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

docs/sql-programming-guide.md

@@ -1490,6 +1490,16 @@ that these options will be deprecated in future release as more optimizations ar
       Configures the number of partitions to use when shuffling data for joins or aggregations.
     </td>
   </tr>
+  <tr>
+    <td><code>spark.sql.typeCoercion.mode</code></td>
+    <td><code>default</code></td>
+    <td>
+      The <code>default</code> type coercion mode was used in spark prior to 2.3.0, and so it
+      continues to be the default to avoid breaking behavior. However, it has logical
+      inconsistencies. The <code>hive</code> mode is preferred for most new applications, though
+      it may require additional manual casting.
+    </td>
+  </tr>
 </table>
 
 ## Broadcast Hint for SQL Queries

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

@@ -151,7 +151,7 @@ class Analyzer(
       TimeWindowing ::
       ResolveInlineTables(conf) ::
       ResolveTimeZone(conf) ::
-      TypeCoercion.typeCoercionRules ++
+      TypeCoercion.rules(conf) ++
       extendedResolutionRules : _*),
     Batch("Post-Hoc Resolution", Once, postHocResolutionRules: _*),
     Batch("View", Once,

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

@@ -27,6 +27,7 @@ import org.apache.spark.sql.catalyst.expressions._
 import org.apache.spark.sql.catalyst.expressions.aggregate._
 import org.apache.spark.sql.catalyst.plans.logical._
 import org.apache.spark.sql.catalyst.rules.Rule
+import org.apache.spark.sql.internal.SQLConf
 import org.apache.spark.sql.types._
 
 
@@ -45,10 +46,8 @@ import org.apache.spark.sql.types._
  */
 object TypeCoercion {
 
-  val typeCoercionRules =
-    InConversion ::
-      WidenSetOperationTypes ::
-      PromoteStrings ::
+  private val commonTypeCoercionRules =
+    WidenSetOperationTypes ::
       DecimalPrecision ::
       BooleanEquality ::
       FunctionArgumentConversion ::
@@ -61,6 +60,18 @@ object TypeCoercion {
       WindowFrameCoercion ::
       Nil
 
+  def rules(conf: SQLConf): List[Rule[LogicalPlan]] = {
+    if (conf.isHiveTypeCoercionMode) {
+      commonTypeCoercionRules :+
+        HiveInConversion :+
+        HivePromoteStrings
+    } else {
+      commonTypeCoercionRules :+
+        InConversion :+
+        PromoteStrings
+    }
+  }
+
   // See https://cwiki.apache.org/confluence/display/Hive/LanguageManual+Types.
   // The conversion for integral and floating point types have a linear widening hierarchy:
   val numericPrecedence =
@@ -148,6 +159,25 @@ object TypeCoercion {
     case (l, r) => None
   }
 
+  val findCommonTypeToCompatibleWithHive: (DataType, DataType) => Option[DataType] = {
+    // Follow hive's binary comparison action:
+    // 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 _ => None
+  }
+
   /**
    * Case 2 type widening (see the classdoc comment above for TypeCoercion).
    *
@@ -309,17 +339,18 @@ object TypeCoercion {
     }
   }
 
+  private def castExpr(expr: Expression, targetType: DataType): Expression = {
+    (expr.dataType, targetType) match {
+      case (NullType, dt) => Literal.create(null, targetType)
+      case (l, dt) if (l != dt) => Cast(expr, targetType)
+      case _ => expr
+    }
+  }
+
   /**
    * Promotes strings that appear in arithmetic expressions.
    */
   object PromoteStrings extends TypeCoercionRule {
-    private def castExpr(expr: Expression, targetType: DataType): Expression = {
-      (expr.dataType, targetType) match {
-        case (NullType, dt) => Literal.create(null, targetType)
-        case (l, dt) if (l != dt) => Cast(expr, targetType)
-        case _ => expr
-      }
-    }
 
     override protected def coerceTypes(plan: LogicalPlan): LogicalPlan = plan resolveExpressions {
       // Skip nodes who's children have not been resolved yet.
@@ -356,6 +387,52 @@ object TypeCoercion {
     }
   }
 
+  /**
+   * Promotes strings that appear in arithmetic expressions to compatible with Hive.
+   */
+  object HivePromoteStrings extends TypeCoercionRule {
+
+    override protected def coerceTypes(plan: LogicalPlan): LogicalPlan = plan resolveExpressions {
+      // Skip nodes who's children have not been resolved yet.
+      case e if !e.childrenResolved => e
+
+      case a @ BinaryArithmetic(left @ StringType(), right) =>
+        a.makeCopy(Array(Cast(left, DoubleType), right))
+      case a @ BinaryArithmetic(left, right @ StringType()) =>
+        a.makeCopy(Array(left, Cast(right, DoubleType)))
+
+      case p @ Equality(left, right)
+        if findCommonTypeToCompatibleWithHive(left.dataType, right.dataType).isDefined =>
+        val commonType = findCommonTypeToCompatibleWithHive(left.dataType, right.dataType).get
+        p.makeCopy(Array(castExpr(left, commonType), castExpr(right, commonType)))
+      case p @ BinaryComparison(left, right)
+        if findCommonTypeToCompatibleWithHive(left.dataType, right.dataType).isDefined =>
+        val commonType = findCommonTypeToCompatibleWithHive(left.dataType, right.dataType).get
+        p.makeCopy(Array(castExpr(left, commonType), castExpr(right, commonType)))
+
+      case Abs(e @ StringType()) => Abs(Cast(e, DoubleType))
+      case Sum(e @ StringType()) => Sum(Cast(e, DoubleType))
+      case Average(e @ StringType()) => Average(Cast(e, DoubleType))
+      case StddevPop(e @ StringType()) => StddevPop(Cast(e, DoubleType))
+      case StddevSamp(e @ StringType()) => StddevSamp(Cast(e, DoubleType))
+      case UnaryMinus(e @ StringType()) => UnaryMinus(Cast(e, DoubleType))
+      case UnaryPositive(e @ StringType()) => UnaryPositive(Cast(e, DoubleType))
+      case VariancePop(e @ StringType()) => VariancePop(Cast(e, DoubleType))
+      case VarianceSamp(e @ StringType()) => VarianceSamp(Cast(e, DoubleType))
+      case Skewness(e @ StringType()) => Skewness(Cast(e, DoubleType))
+      case Kurtosis(e @ StringType()) => Kurtosis(Cast(e, DoubleType))
+    }
+  }
+
+  private def flattenExpr(expr: Expression): Seq[Expression] = {
+    expr match {
+      // Multi columns in IN clause is represented as a CreateNamedStruct.
+      // flatten the named struct to get the list of expressions.
+      case cns: CreateNamedStruct => cns.valExprs
+      case expr => Seq(expr)
+    }
+  }
+
   /**
    * Handles type coercion for both IN expression with subquery and IN
    * expressions without subquery.
@@ -371,14 +448,6 @@ object TypeCoercion {
    *    Analysis Exception will be raised at the type checking phase.
    */
   object InConversion extends TypeCoercionRule {
-    private def flattenExpr(expr: Expression): Seq[Expression] = {
-      expr match {
-        // Multi columns in IN clause is represented as a CreateNamedStruct.
-        // flatten the named struct to get the list of expressions.
-        case cns: CreateNamedStruct => cns.valExprs
-        case expr => Seq(expr)
-      }
-    }
 
     override protected def coerceTypes(plan: LogicalPlan): LogicalPlan = plan resolveExpressions {
       // Skip nodes who's children have not been resolved yet.
@@ -432,6 +501,64 @@ object TypeCoercion {
     }
   }
 
+  /**
+   * Handles type coercion for IN expression to compatible with Hive.
+   */
+  object HiveInConversion extends TypeCoercionRule {
+
+    override protected def coerceTypes(plan: LogicalPlan): LogicalPlan = plan resolveExpressions {
+      // Skip nodes who's children have not been resolved yet.
+      case e if !e.childrenResolved => e
+
+      // Handle type casting required between value expression and subquery output
+      // in IN subquery.
+      case i @ In(a, Seq(ListQuery(sub, children, exprId, _)))
+        if !i.resolved && flattenExpr(a).length == sub.output.length =>
+        // LHS is the value expression of IN subquery.
+        val lhs = flattenExpr(a)
+
+        // RHS is the subquery output.
+        val rhs = sub.output
+
+        val commonTypes = lhs.zip(rhs).flatMap { case (l, r) =>
+          findCommonTypeToCompatibleWithHive(l.dataType, r.dataType)
+            .orElse(findTightestCommonType(l.dataType, r.dataType))
+
+        }
+
+        // The number of columns/expressions must match between LHS and RHS of an
+        // IN subquery expression.
+        if (commonTypes.length == lhs.length) {
+          val castedRhs = rhs.zip(commonTypes).map {
+            case (e, dt) if e.dataType != dt => Alias(Cast(e, dt), e.name)()
+            case (e, _) => e
+          }
+          val castedLhs = lhs.zip(commonTypes).map {
+            case (e, dt) if e.dataType != dt => Cast(e, dt)
+            case (e, _) => e
+          }
+
+          // Before constructing the In expression, wrap the multi values in LHS
+          // in a CreatedNamedStruct.
+          val newLhs = castedLhs match {
+            case Seq(lhs) => lhs
+            case _ => CreateStruct(castedLhs)
+          }
+
+          val newSub = Project(castedRhs, sub)
+          In(newLhs, Seq(ListQuery(newSub, children, exprId, newSub.output)))
+        } else {
+          i
+        }
+
+      case i @ In(a, b) if b.exists(_.dataType != a.dataType) =>
+        findWiderCommonType(i.children.map(_.dataType)) match {
+          case Some(finalDataType) => i.withNewChildren(i.children.map(Cast(_, finalDataType)))
+          case None => i
+        }
+    }
+  }
+
   /**
    * Changes numeric values to booleans so that expressions like true = 1 can be evaluated.
    */

sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala

@@ -1015,6 +1015,18 @@ object SQLConf {
       .booleanConf
       .createWithDefault(true)
 
+  val typeCoercionMode =
+    buildConf("spark.sql.typeCoercion.mode")
+      .doc("The 'default' typeCoercion mode was used in spark prior to 2.3, " +
+        "and so it continues to be the default to avoid breaking behavior. " +
+        "However, it has logical inconsistencies. " +
+        "The 'hive' mode is preferred for most new applications, " +
+        "though it may require additional manual casting.")
+      .stringConf
+      .transform(_.toLowerCase(Locale.ROOT))
+      .checkValues(Set("default", "hive"))
+      .createWithDefault("default")
+
   val REPLACE_EXCEPT_WITH_FILTER = buildConf("spark.sql.optimizer.replaceExceptWithFilter")
     .internal()
     .doc("When true, the apply function of the rule verifies whether the right node of the" +
@@ -1335,6 +1347,8 @@ class SQLConf extends Serializable with Logging {
 
   def pandasRespectSessionTimeZone: Boolean = getConf(PANDAS_RESPECT_SESSION_LOCAL_TIMEZONE)
 
+  def isHiveTypeCoercionMode: Boolean = getConf(SQLConf.typeCoercionMode).equals("hive")
+
   def replaceExceptWithFilter: Boolean = getConf(REPLACE_EXCEPT_WITH_FILTER)
 
   /** ********************** SQLConf functionality methods ************ */