Improve error reporting for expression data type mismatch

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

@@ -62,15 +62,15 @@ trait CheckAnalysis {
             val from = operator.inputSet.map(_.name).mkString(", ")
             a.failAnalysis(s"cannot resolve '${a.prettyString}' given input columns $from")
 
+          case e: Expression if !e.validInputTypes =>
+            e.failAnalysis(
+              s"cannot resolve '${t.prettyString}' due to data type mismatch: " +
+              e.typeMismatchErrorMessage.get)
+
           case c: Cast if !c.resolved =>
             failAnalysis(
               s"invalid cast from ${c.child.dataType.simpleString} to ${c.dataType.simpleString}")
 
-          case b: BinaryExpression if !b.resolved =>
-            failAnalysis(
-              s"invalid expression ${b.prettyString} " +
-              s"between ${b.left.dataType.simpleString} and ${b.right.dataType.simpleString}")
-
           case WindowExpression(UnresolvedWindowFunction(name, _), _) =>
             failAnalysis(
               s"Could not resolve window function '$name'. " +

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

@@ -407,7 +407,7 @@ trait HiveTypeCoercion {
         Union(newLeft, newRight)
 
       // fix decimal precision for expressions
-      case q => q.transformExpressions {
+      case q => q.transformExpressionsUp {
         // Skip nodes whose children have not been resolved yet
         case e if !e.childrenResolved => e
 

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

@@ -86,12 +86,16 @@ abstract class Expression extends TreeNode[Expression] {
       case (i1, i2) => i1 == i2
     }
   }
+
+  def typeMismatchErrorMessage: Option[String] = None
+
+  def validInputTypes: Boolean = typeMismatchErrorMessage.isEmpty
 }
 
 abstract class BinaryExpression extends Expression with trees.BinaryNode[Expression] {
   self: Product =>
 
-  def symbol: String
+  def symbol: String = sys.error(s"BinaryExpressions must either override toString or symbol")
 
   override def foldable: Boolean = left.foldable && right.foldable
 
@@ -106,6 +110,10 @@ abstract class LeafExpression extends Expression with trees.LeafNode[Expression]
 
 abstract class UnaryExpression extends Expression with trees.UnaryNode[Expression] {
   self: Product =>
+
+  override def foldable: Boolean = child.foldable
+
+  override def nullable: Boolean = child.nullable
 }
 
 // TODO Semantically we probably not need GroupExpression
@@ -125,7 +133,9 @@ case class GroupExpression(children: Seq[Expression]) extends Expression {
  * so that the proper type conversions can be performed in the analyzer.
  */
 trait ExpectsInputTypes {
+  self: Expression =>
 
   def expectedChildTypes: Seq[DataType]
 
+  override def validInputTypes: Boolean = children.map(_.dataType) == expectedChildTypes
 }