[SPARK-18207][SQL] Fix a compilation error due to HashExpression.doGenCode

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

@@ -268,15 +268,16 @@ abstract class HashExpression[E] extends Expression {
 
   override def doGenCode(ctx: CodegenContext, ev: ExprCode): ExprCode = {
     ev.isNull = "false"
-    val childrenHash = children.map { child =>
+    val childrenHash = ctx.splitExpressions(ctx.INPUT_ROW, children.map { child =>
       val childGen = child.genCode(ctx)
       childGen.code + ctx.nullSafeExec(child.nullable, childGen.isNull) {
         computeHash(childGen.value, child.dataType, ev.value, ctx)
       }
-    }.mkString("\n")
+    })
 
+    ctx.addMutableState(ctx.javaType(dataType), ev.value, "")
     ev.copy(code = s"""
-      ${ctx.javaType(dataType)} ${ev.value} = $seed;
+      ${ev.value} = $seed;
       $childrenHash""")
   }
 
@@ -600,15 +601,18 @@ case class HiveHash(children: Seq[Expression]) extends HashExpression[Int] {
   override def doGenCode(ctx: CodegenContext, ev: ExprCode): ExprCode = {
     ev.isNull = "false"
     val childHash = ctx.freshName("childHash")
-    val childrenHash = children.map { child =>
+    val childrenHash = ctx.splitExpressions(ctx.INPUT_ROW, children.map { child =>
       val childGen = child.genCode(ctx)
       childGen.code + ctx.nullSafeExec(child.nullable, childGen.isNull) {
         computeHash(childGen.value, child.dataType, childHash, ctx)
-      } + s"${ev.value} = (31 * ${ev.value}) + $childHash;"
-    }.mkString(s"int $childHash = 0;", s"\n$childHash = 0;\n", "")
+      } + s"${ev.value} = (31 * ${ev.value}) + $childHash;" +
+        s"\n$childHash = 0;"
+    })
 
+    ctx.addMutableState(ctx.javaType(dataType), ev.value, "")
+    ctx.addMutableState("int", childHash, s"$childHash = 0;")
     ev.copy(code = s"""
-      ${ctx.javaType(dataType)} ${ev.value} = $seed;
+      ${ev.value} = $seed;
       $childrenHash""")
   }
 

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

@@ -1728,4 +1728,23 @@ class DataFrameSuite extends QueryTest with SharedSQLContext {
     val df = spark.createDataFrame(spark.sparkContext.makeRDD(rows), schema)
     assert(df.filter($"array1" === $"array2").count() == 1)
   }
+
+  test("SPARK-18207: Compute hash for wider table") {
+    import org.apache.spark.sql.types.{StructType, StringType}
+
+    val COLMAX = 1000
+    val schema: StructType = (1 to COLMAX)
+      .foldLeft(new StructType())((s, i) => s.add(s"g$i", StringType, nullable = true))
+    val rdds = spark.sparkContext.parallelize(Seq(Row.fromSeq((1 to COLMAX).map(_.toString))))
+    val wideDF = spark.createDataFrame(rdds, schema)
+
+    val widePlus = wideDF.withColumn("d_rank", lit(1))
+    widePlus.createOrReplaceTempView("wide_plus")
+    val widePlus2 = widePlus.withColumn("d_rank", lit(0))
+    widePlus2.createOrReplaceTempView("wide_plus2")
+
+    // union operation in this SQL involves computation of hash for a row
+    val df = spark.sqlContext.sql("select * from wide_plus union select * from wide_plus2")
+    df.count
+  }
 }