BinaryOperator should not be doing implicit cast.

Author: rxin

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

@@ -221,10 +221,16 @@ object HiveTypeCoercion {
         case e if !e.childrenResolved => e
 
         case b @ BinaryOperator(left, right) if left.dataType != right.dataType =>
-          findTightestCommonTypeOfTwo(left.dataType, right.dataType).map { widestType =>
-            val newLeft = if (left.dataType == widestType) left else Cast(left, widestType)
-            val newRight = if (right.dataType == widestType) right else Cast(right, widestType)
-            b.makeCopy(Array(newLeft, newRight))
+          findTightestCommonTypeOfTwo(left.dataType, right.dataType).map { commonType =>
+            // If the expression accepts the tighest common type, cast to that.
+            // Otherwise, don't do anything with the expression.
+            if (b.inputType.acceptsType(commonType)) {
+              val newLeft = if (left.dataType == commonType) left else Cast(left, commonType)
+              val newRight = if (right.dataType == commonType) right else Cast(right, commonType)
+              b.makeCopy(Array(newLeft, newRight))
+            } else {
+              b
+            }
           }.getOrElse(b)  // If there is no applicable conversion, leave expression unchanged.
       }
     }
@@ -680,7 +686,7 @@ object HiveTypeCoercion {
       // Skip nodes who's children have not been resolved yet.
       case e if !e.childrenResolved => e
 
-      case e: ExpectsInputTypes if (e.inputTypes.nonEmpty) =>
+      case e: ExpectsInputTypes if e.inputTypes.nonEmpty =>
         val children: Seq[Expression] = e.children.zip(e.inputTypes).map { case (in, expected) =>
           // If we cannot do the implicit cast, just use the original input.
           implicitCast(in, expected).getOrElse(in)

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

@@ -356,20 +356,32 @@ abstract class BinaryExpression extends Expression with trees.BinaryNode[Express
 abstract class BinaryOperator extends BinaryExpression {
   self: Product =>
 
+  /**
+   * Expected input type from both left/right child expressions, similar to the
+   * [[ExpectsInputTypes]] trait.
+   */
+  def inputType: AbstractDataType
+
   def symbol: String
 
+  override def toString: String = s"($left $symbol $right)"
+
   override def checkInputDataTypes(): TypeCheckResult = {
-    // First call the checker for ExpectsInputTypes, and then check whether left and right have
-    // the same type.
-    super.checkInputDataTypes() match {
-      case TypeCheckResult.TypeCheckSuccess =>
-        if (left.dataType != right.dataType) {
-          TypeCheckResult.TypeCheckFailure(s"differing types in '$prettyString' " +
-            s"(${left.dataType.simpleString} and ${right.dataType.simpleString}).")
-        } else {
-          TypeCheckResult.TypeCheckSuccess
-        }
-      case TypeCheckResult.TypeCheckFailure(msg) => TypeCheckResult.TypeCheckFailure(msg)
+    val mismatches = children.zipWithIndex.collect {
+      case (child, idx) if !inputType.acceptsType(child.dataType) =>
+        s"argument ${idx + 1} is expected to be of type ${inputType.simpleString}, " +
+          s"however, '${child.prettyString}' is of type ${child.dataType.simpleString}."
+    }
+
+    if (mismatches.isEmpty) {
+      if (left.dataType != right.dataType) {
+        TypeCheckResult.TypeCheckFailure(s"differing types in '$prettyString' " +
+          s"(${left.dataType.simpleString} and ${right.dataType.simpleString}).")
+      } else {
+        TypeCheckResult.TypeCheckSuccess
+      }
+    } else {
+      TypeCheckResult.TypeCheckFailure(mismatches.mkString(" "))
     }
   }
 }

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

@@ -29,12 +29,13 @@ abstract class UnaryArithmetic extends UnaryExpression {
   override def dataType: DataType = child.dataType
 }
 
-case class UnaryMinus(child: Expression) extends UnaryArithmetic with ExpectsInputTypes {
-
-  override def inputTypes: Seq[AbstractDataType] = Seq(NumericType)
+case class UnaryMinus(child: Expression) extends UnaryArithmetic {
 
   override def toString: String = s"-$child"
 
+  override def checkInputDataTypes(): TypeCheckResult =
+    TypeUtils.checkForNumericExpr(child.dataType, "operator -")
+
   private lazy val numeric = TypeUtils.getNumeric(dataType)
 
   override def genCode(ctx: CodeGenContext, ev: GeneratedExpressionCode): String = dataType match {
@@ -48,6 +49,9 @@ case class UnaryMinus(child: Expression) extends UnaryArithmetic with ExpectsInp
 case class UnaryPositive(child: Expression) extends UnaryArithmetic {
   override def prettyName: String = "positive"
 
+  override def checkInputDataTypes(): TypeCheckResult =
+    TypeUtils.checkForNumericExpr(child.dataType, "operator -")
+
   override def genCode(ctx: CodeGenContext, ev: GeneratedExpressionCode): String =
     defineCodeGen(ctx, ev, c => c)
 
@@ -57,9 +61,9 @@ case class UnaryPositive(child: Expression) extends UnaryArithmetic {
 /**
  * A function that get the absolute value of the numeric value.
  */
-case class Abs(child: Expression) extends UnaryArithmetic with ExpectsInputTypes {
-
-  override def inputTypes: Seq[AbstractDataType] = Seq(NumericType)
+case class Abs(child: Expression) extends UnaryArithmetic {
+  override def checkInputDataTypes(): TypeCheckResult =
+    TypeUtils.checkForNumericExpr(child.dataType, "function abs")
 
   private lazy val numeric = TypeUtils.getNumeric(dataType)
 
@@ -91,14 +95,13 @@ private[sql] object BinaryArithmetic {
   def unapply(e: BinaryArithmetic): Option[(Expression, Expression)] = Some((e.left, e.right))
 }
 
-case class Add(left: Expression, right: Expression)
-  extends BinaryArithmetic with ExpectsInputTypes {
+case class Add(left: Expression, right: Expression) extends BinaryArithmetic {
+
+  override def inputType: AbstractDataType = NumericType
 
   override def symbol: String = "+"
   override def decimalMethod: String = "$plus"
 
-  override def inputTypes: Seq[AbstractDataType] = Seq(NumericType, NumericType)
-
   override lazy val resolved =
     childrenResolved && checkInputDataTypes().isSuccess && !DecimalType.isFixed(dataType)
 
@@ -107,14 +110,13 @@ case class Add(left: Expression, right: Expression)
   protected override def nullSafeEval(input1: Any, input2: Any): Any = numeric.plus(input1, input2)
 }
 
-case class Subtract(left: Expression, right: Expression)
-  extends BinaryArithmetic with ExpectsInputTypes {
+case class Subtract(left: Expression, right: Expression) extends BinaryArithmetic {
+
+  override def inputType: AbstractDataType = NumericType
 
   override def symbol: String = "-"
   override def decimalMethod: String = "$minus"
 
-  override def inputTypes: Seq[AbstractDataType] = Seq(NumericType, NumericType)
-
   override lazy val resolved =
     childrenResolved && checkInputDataTypes().isSuccess && !DecimalType.isFixed(dataType)
 
@@ -123,14 +125,13 @@ case class Subtract(left: Expression, right: Expression)
   protected override def nullSafeEval(input1: Any, input2: Any): Any = numeric.minus(input1, input2)
 }
 
-case class Multiply(left: Expression, right: Expression)
-  extends BinaryArithmetic with ExpectsInputTypes {
+case class Multiply(left: Expression, right: Expression) extends BinaryArithmetic {
+
+  override def inputType: AbstractDataType = NumericType
 
   override def symbol: String = "*"
   override def decimalMethod: String = "$times"
 
-  override def inputTypes: Seq[AbstractDataType] = Seq(NumericType, NumericType)
-
   override lazy val resolved =
     childrenResolved && checkInputDataTypes().isSuccess && !DecimalType.isFixed(dataType)
 
@@ -139,15 +140,14 @@ case class Multiply(left: Expression, right: Expression)
   protected override def nullSafeEval(input1: Any, input2: Any): Any = numeric.times(input1, input2)
 }
 
-case class Divide(left: Expression, right: Expression)
-  extends BinaryArithmetic with ExpectsInputTypes {
+case class Divide(left: Expression, right: Expression) extends BinaryArithmetic {
+
+  override def inputType: AbstractDataType = NumericType
 
   override def symbol: String = "/"
   override def decimalMethod: String = "$div"
   override def nullable: Boolean = true
 
-  override def inputTypes: Seq[AbstractDataType] = Seq(NumericType, NumericType)
-
   override lazy val resolved =
     childrenResolved && checkInputDataTypes().isSuccess && !DecimalType.isFixed(dataType)
 
@@ -205,15 +205,14 @@ case class Divide(left: Expression, right: Expression)
   }
 }
 
-case class Remainder(left: Expression, right: Expression)
-  extends BinaryArithmetic with ExpectsInputTypes {
+case class Remainder(left: Expression, right: Expression) extends BinaryArithmetic {
+
+  override def inputType: AbstractDataType = NumericType
 
   override def symbol: String = "%"
   override def decimalMethod: String = "remainder"
   override def nullable: Boolean = true
 
-  override def inputTypes: Seq[AbstractDataType] = Seq(NumericType, NumericType)
-
   override lazy val resolved =
     childrenResolved && checkInputDataTypes().isSuccess && !DecimalType.isFixed(dataType)
 
@@ -272,6 +271,10 @@ case class Remainder(left: Expression, right: Expression)
 }
 
 case class MaxOf(left: Expression, right: Expression) extends BinaryArithmetic {
+  // TODO: Remove MaxOf and MinOf, and replace its usage with Greatest and Least.
+
+  override def inputType: AbstractDataType = TypeCollection.Ordered
+
   override def nullable: Boolean = left.nullable && right.nullable
 
   protected def checkTypesInternal(t: DataType) =
@@ -326,6 +329,10 @@ case class MaxOf(left: Expression, right: Expression) extends BinaryArithmetic {
 }
 
 case class MinOf(left: Expression, right: Expression) extends BinaryArithmetic {
+  // TODO: Remove MaxOf and MinOf, and replace its usage with Greatest and Least.
+
+  override def inputType: AbstractDataType = TypeCollection.Ordered
+
   override def nullable: Boolean = left.nullable && right.nullable
 
   protected def checkTypesInternal(t: DataType) =

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

@@ -17,7 +17,9 @@
 
 package org.apache.spark.sql.catalyst.expressions
 
+import org.apache.spark.sql.catalyst.analysis.TypeCheckResult
 import org.apache.spark.sql.catalyst.expressions.codegen._
+import org.apache.spark.sql.catalyst.util.TypeUtils
 import org.apache.spark.sql.types._
 
 
@@ -26,13 +28,11 @@ import org.apache.spark.sql.types._
  *
  * Code generation inherited from BinaryArithmetic.
  */
-case class BitwiseAnd(left: Expression, right: Expression)
-  extends BinaryArithmetic with ExpectsInputTypes {
+case class BitwiseAnd(left: Expression, right: Expression) extends BinaryArithmetic {
 
-  override def symbol: String = "&"
+  override def inputType: AbstractDataType = TypeCollection.Bitwise
 
-  override def inputTypes: Seq[AbstractDataType] =
-    Seq(TypeCollection(IntegerType, LongType), TypeCollection(IntegerType, LongType))
+  override def symbol: String = "&"
 
   private lazy val and: (Any, Any) => Any = dataType match {
     case ByteType =>
@@ -53,13 +53,11 @@ case class BitwiseAnd(left: Expression, right: Expression)
  *
  * Code generation inherited from BinaryArithmetic.
  */
-case class BitwiseOr(left: Expression, right: Expression)
-  extends BinaryArithmetic with ExpectsInputTypes {
+case class BitwiseOr(left: Expression, right: Expression) extends BinaryArithmetic {
 
-  override def symbol: String = "|"
+  override def inputType: AbstractDataType = TypeCollection.Bitwise
 
-  override def inputTypes: Seq[AbstractDataType] =
-    Seq(TypeCollection(IntegerType, LongType), TypeCollection(IntegerType, LongType))
+  override def symbol: String = "|"
 
   private lazy val or: (Any, Any) => Any = dataType match {
     case ByteType =>
@@ -80,13 +78,11 @@ case class BitwiseOr(left: Expression, right: Expression)
  *
  * Code generation inherited from BinaryArithmetic.
  */
-case class BitwiseXor(left: Expression, right: Expression)
-  extends BinaryArithmetic with ExpectsInputTypes {
+case class BitwiseXor(left: Expression, right: Expression) extends BinaryArithmetic {
 
-  override def symbol: String = "^"
+  override def inputType: AbstractDataType = TypeCollection.Bitwise
 
-  override def inputTypes: Seq[AbstractDataType] =
-    Seq(TypeCollection(IntegerType, LongType), TypeCollection(IntegerType, LongType))
+  override def symbol: String = "^"
 
   private lazy val xor: (Any, Any) => Any = dataType match {
     case ByteType =>
@@ -105,11 +101,12 @@ case class BitwiseXor(left: Expression, right: Expression)
 /**
  * A function that calculates bitwise not(~) of a number.
  */
-case class BitwiseNot(child: Expression) extends UnaryArithmetic with ExpectsInputTypes {
+case class BitwiseNot(child: Expression) extends UnaryArithmetic {
 
-  override def toString: String = s"~$child"
+  override def checkInputDataTypes(): TypeCheckResult =
+    TypeUtils.checkForBitwiseExpr(child.dataType, "operator ~")
 
-  override def inputTypes: Seq[AbstractDataType] = Seq(TypeCollection(IntegerType, LongType))
+  override def toString: String = s"~$child"
 
   private lazy val not: (Any) => Any = dataType match {
     case ByteType =>