From de7249a92b331eeee97061e59cd52b0deb85bae5 Mon Sep 17 00:00:00 2001 From: "Jungtaek Lim (HeartSaVioR)" Date: Wed, 18 Sep 2019 21:48:38 +0900 Subject: [PATCH 1/5] [SPARK-29140][SQL] Handle BinaryType of parameter properly in HashAggregateExec --- .../aggregate/HashAggregateExec.scala | 12 ++++- .../aggregate/HashAggregateSuite.scala | 47 +++++++++++++++++++ 2 files changed, 58 insertions(+), 1 deletion(-) create mode 100644 sql/core/src/test/scala/org/apache/spark/sql/execution/aggregate/HashAggregateSuite.scala diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/aggregate/HashAggregateExec.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/aggregate/HashAggregateExec.scala index 2d187e3c9ebe5..330581855c896 100644 --- a/sql/core/src/main/scala/org/apache/spark/sql/execution/aggregate/HashAggregateExec.scala +++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/aggregate/HashAggregateExec.scala @@ -299,7 +299,9 @@ case class HashAggregateExec( if (inputVars.forall(_.isDefined)) { val splitCodes = inputVars.flatten.zipWithIndex.map { case (args, i) => val doAggFunc = ctx.freshName(s"doAggregate_${aggNames(i)}") - val argList = args.map(v => s"${v.javaType.getName} ${v.variableName}").mkString(", ") + val argList = args.map { v => + s"${typeNameForCodegen(v.javaType)} ${v.variableName}" + }.mkString(", ") val doAggFuncName = ctx.addNewFunction(doAggFunc, s""" |private void $doAggFunc($argList) throws java.io.IOException { @@ -392,6 +394,14 @@ case class HashAggregateExec( """.stripMargin } + private def typeNameForCodegen(clazz: Class[_]): String = { + if (clazz.isArray) { + typeNameForCodegen(clazz.getComponentType) + "[]" + } else { + clazz.getName + } + } + private val groupingAttributes = groupingExpressions.map(_.toAttribute) private val groupingKeySchema = StructType.fromAttributes(groupingAttributes) private val declFunctions = aggregateExpressions.map(_.aggregateFunction) diff --git a/sql/core/src/test/scala/org/apache/spark/sql/execution/aggregate/HashAggregateSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/execution/aggregate/HashAggregateSuite.scala new file mode 100644 index 0000000000000..28d30fc5c3dce --- /dev/null +++ b/sql/core/src/test/scala/org/apache/spark/sql/execution/aggregate/HashAggregateSuite.scala @@ -0,0 +1,47 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.spark.sql.execution.aggregate + +import org.apache.spark.sql.Row +import org.apache.spark.sql.functions._ +import org.apache.spark.sql.internal.SQLConf +import org.apache.spark.sql.test.SharedSparkSession +import org.apache.spark.sql.types._ + +class HashAggregateSuite extends SharedSparkSession { + + import testImplicits._ + + test("SPARK-29140 HashAggregateExec aggregating binary type doesn't break codegen compilation") { + val withDistinct = countDistinct($"c1") + + val schema = new StructType().add("c1", BinaryType, nullable = true) + val schemaWithId = StructType(StructField("id", IntegerType, nullable = false) +: schema.fields) + + withSQLConf( + SQLConf.CODEGEN_SPLIT_AGGREGATE_FUNC.key -> "true", + SQLConf.CODEGEN_METHOD_SPLIT_THRESHOLD.key -> "1") { + val emptyRows = spark.sparkContext.parallelize(Seq.empty[Row], 1) + val aggDf = spark.createDataFrame(emptyRows, schemaWithId) + .groupBy($"id" % 10 as "group") + .agg(withDistinct) + .orderBy("group") + aggDf.collect().toSeq + } + } +} From 9ce98b25c282fc1c4519ec17887af2f01d09672a Mon Sep 17 00:00:00 2001 From: "Jungtaek Lim (HeartSaVioR)" Date: Thu, 19 Sep 2019 05:53:27 +0900 Subject: [PATCH 2/5] Reflect review comments --- .../expressions/codegen/CodeGenerator.scala | 8 ++++ .../aggregate/HashAggregateExec.scala | 10 +--- .../aggregate/HashAggregateSuite.scala | 47 ------------------- .../execution/AggregationQuerySuite.scala | 20 +++++++- 4 files changed, 28 insertions(+), 57 deletions(-) delete mode 100644 sql/core/src/test/scala/org/apache/spark/sql/execution/aggregate/HashAggregateSuite.scala diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/codegen/CodeGenerator.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/codegen/CodeGenerator.scala index 68ddec9fc8d00..946fc7f421ad6 100644 --- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/codegen/CodeGenerator.scala +++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/codegen/CodeGenerator.scala @@ -1811,6 +1811,14 @@ object CodeGenerator extends Logging { def boxedType(dt: DataType): String = boxedType(javaType(dt)) + def typeName(clazz: Class[_]): String = { + if (clazz.isArray) { + typeName(clazz.getComponentType) + "[]" + } else { + clazz.getName + } + } + /** * Returns the representation of default value for a given Java Type. * @param jt the string name of the Java type diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/aggregate/HashAggregateExec.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/aggregate/HashAggregateExec.scala index 330581855c896..5dc5b822919be 100644 --- a/sql/core/src/main/scala/org/apache/spark/sql/execution/aggregate/HashAggregateExec.scala +++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/aggregate/HashAggregateExec.scala @@ -300,7 +300,7 @@ case class HashAggregateExec( val splitCodes = inputVars.flatten.zipWithIndex.map { case (args, i) => val doAggFunc = ctx.freshName(s"doAggregate_${aggNames(i)}") val argList = args.map { v => - s"${typeNameForCodegen(v.javaType)} ${v.variableName}" + s"${CodeGenerator.typeName(v.javaType)} ${v.variableName}" }.mkString(", ") val doAggFuncName = ctx.addNewFunction(doAggFunc, s""" @@ -394,14 +394,6 @@ case class HashAggregateExec( """.stripMargin } - private def typeNameForCodegen(clazz: Class[_]): String = { - if (clazz.isArray) { - typeNameForCodegen(clazz.getComponentType) + "[]" - } else { - clazz.getName - } - } - private val groupingAttributes = groupingExpressions.map(_.toAttribute) private val groupingKeySchema = StructType.fromAttributes(groupingAttributes) private val declFunctions = aggregateExpressions.map(_.aggregateFunction) diff --git a/sql/core/src/test/scala/org/apache/spark/sql/execution/aggregate/HashAggregateSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/execution/aggregate/HashAggregateSuite.scala deleted file mode 100644 index 28d30fc5c3dce..0000000000000 --- a/sql/core/src/test/scala/org/apache/spark/sql/execution/aggregate/HashAggregateSuite.scala +++ /dev/null @@ -1,47 +0,0 @@ -/* - * Licensed to the Apache Software Foundation (ASF) under one or more - * contributor license agreements. See the NOTICE file distributed with - * this work for additional information regarding copyright ownership. - * The ASF licenses this file to You under the Apache License, Version 2.0 - * (the "License"); you may not use this file except in compliance with - * the License. You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ - -package org.apache.spark.sql.execution.aggregate - -import org.apache.spark.sql.Row -import org.apache.spark.sql.functions._ -import org.apache.spark.sql.internal.SQLConf -import org.apache.spark.sql.test.SharedSparkSession -import org.apache.spark.sql.types._ - -class HashAggregateSuite extends SharedSparkSession { - - import testImplicits._ - - test("SPARK-29140 HashAggregateExec aggregating binary type doesn't break codegen compilation") { - val withDistinct = countDistinct($"c1") - - val schema = new StructType().add("c1", BinaryType, nullable = true) - val schemaWithId = StructType(StructField("id", IntegerType, nullable = false) +: schema.fields) - - withSQLConf( - SQLConf.CODEGEN_SPLIT_AGGREGATE_FUNC.key -> "true", - SQLConf.CODEGEN_METHOD_SPLIT_THRESHOLD.key -> "1") { - val emptyRows = spark.sparkContext.parallelize(Seq.empty[Row], 1) - val aggDf = spark.createDataFrame(emptyRows, schemaWithId) - .groupBy($"id" % 10 as "group") - .agg(withDistinct) - .orderBy("group") - aggDf.collect().toSeq - } - } -} diff --git a/sql/hive/src/test/scala/org/apache/spark/sql/hive/execution/AggregationQuerySuite.scala b/sql/hive/src/test/scala/org/apache/spark/sql/hive/execution/AggregationQuerySuite.scala index 8c7e5bf5ac1d4..f6eaed2ecae4e 100644 --- a/sql/hive/src/test/scala/org/apache/spark/sql/hive/execution/AggregationQuerySuite.scala +++ b/sql/hive/src/test/scala/org/apache/spark/sql/hive/execution/AggregationQuerySuite.scala @@ -1021,13 +1021,31 @@ abstract class AggregationQuerySuite extends QueryTest with SQLTestUtils with Te test("SPARK-29122: hash-based aggregates for unfixed-length decimals in the interpreter mode") { withSQLConf(SQLConf.WHOLESTAGE_CODEGEN_ENABLED.key -> "false", - SQLConf.CODEGEN_FACTORY_MODE.key -> CodegenObjectFactoryMode.NO_CODEGEN.toString) { + SQLConf.CODEGEN_FACTORY_MODE.key -> CodegenObjectFactoryMode.NO_CODEGEN.toString) { withTempView("t") { spark.range(3).selectExpr("CAST(id AS decimal(38, 0)) a").createOrReplaceTempView("t") checkAnswer(sql("SELECT SUM(a) FROM t"), Row(java.math.BigDecimal.valueOf(3))) } } } + + test("SPARK-29140: HashAggregateExec aggregating binary type doesn't break codegen compilation") { + val withDistinct = countDistinct($"c1") + + val schema = new StructType().add("c1", BinaryType, nullable = true) + val schemaWithId = StructType(StructField("id", IntegerType, nullable = false) +: schema.fields) + + withSQLConf( + SQLConf.CODEGEN_SPLIT_AGGREGATE_FUNC.key -> "true", + SQLConf.CODEGEN_METHOD_SPLIT_THRESHOLD.key -> "1") { + val emptyRows = spark.sparkContext.parallelize(Seq.empty[Row], 1) + val aggDf = spark.createDataFrame(emptyRows, schemaWithId) + .groupBy($"id" % 10 as "group") + .agg(withDistinct) + .orderBy("group") + checkAnswer(aggDf, Seq.empty[Row]) + } + } } From f045457de14d8c4e16f6b2c1158d167dbb677122 Mon Sep 17 00:00:00 2001 From: "Jungtaek Lim (HeartSaVioR)" Date: Thu, 19 Sep 2019 06:54:33 +0900 Subject: [PATCH 3/5] Additional review comment for nits --- .../spark/sql/hive/execution/AggregationQuerySuite.scala | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/sql/hive/src/test/scala/org/apache/spark/sql/hive/execution/AggregationQuerySuite.scala b/sql/hive/src/test/scala/org/apache/spark/sql/hive/execution/AggregationQuerySuite.scala index f6eaed2ecae4e..cf41a3b97d9cf 100644 --- a/sql/hive/src/test/scala/org/apache/spark/sql/hive/execution/AggregationQuerySuite.scala +++ b/sql/hive/src/test/scala/org/apache/spark/sql/hive/execution/AggregationQuerySuite.scala @@ -1032,14 +1032,14 @@ abstract class AggregationQuerySuite extends QueryTest with SQLTestUtils with Te test("SPARK-29140: HashAggregateExec aggregating binary type doesn't break codegen compilation") { val withDistinct = countDistinct($"c1") - val schema = new StructType().add("c1", BinaryType, nullable = true) - val schemaWithId = StructType(StructField("id", IntegerType, nullable = false) +: schema.fields) + val schema = new StructType().add("id", IntegerType, nullable = false) + .add("c1", BinaryType, nullable = true) withSQLConf( SQLConf.CODEGEN_SPLIT_AGGREGATE_FUNC.key -> "true", SQLConf.CODEGEN_METHOD_SPLIT_THRESHOLD.key -> "1") { val emptyRows = spark.sparkContext.parallelize(Seq.empty[Row], 1) - val aggDf = spark.createDataFrame(emptyRows, schemaWithId) + val aggDf = spark.createDataFrame(emptyRows, schema) .groupBy($"id" % 10 as "group") .agg(withDistinct) .orderBy("group") From 28726dab8ec7e4e4b7bffc51e8feb5481fd12802 Mon Sep 17 00:00:00 2001 From: "Jungtaek Lim (HeartSaVioR)" Date: Fri, 20 Sep 2019 17:56:03 +0900 Subject: [PATCH 4/5] Reflect review comments --- .../spark/sql/hive/execution/AggregationQuerySuite.scala | 5 +---- 1 file changed, 1 insertion(+), 4 deletions(-) diff --git a/sql/hive/src/test/scala/org/apache/spark/sql/hive/execution/AggregationQuerySuite.scala b/sql/hive/src/test/scala/org/apache/spark/sql/hive/execution/AggregationQuerySuite.scala index cf41a3b97d9cf..86e1aeca9f6d6 100644 --- a/sql/hive/src/test/scala/org/apache/spark/sql/hive/execution/AggregationQuerySuite.scala +++ b/sql/hive/src/test/scala/org/apache/spark/sql/hive/execution/AggregationQuerySuite.scala @@ -1030,8 +1030,6 @@ abstract class AggregationQuerySuite extends QueryTest with SQLTestUtils with Te } test("SPARK-29140: HashAggregateExec aggregating binary type doesn't break codegen compilation") { - val withDistinct = countDistinct($"c1") - val schema = new StructType().add("id", IntegerType, nullable = false) .add("c1", BinaryType, nullable = true) @@ -1041,8 +1039,7 @@ abstract class AggregationQuerySuite extends QueryTest with SQLTestUtils with Te val emptyRows = spark.sparkContext.parallelize(Seq.empty[Row], 1) val aggDf = spark.createDataFrame(emptyRows, schema) .groupBy($"id" % 10 as "group") - .agg(withDistinct) - .orderBy("group") + .agg(countDistinct($"c1")) checkAnswer(aggDf, Seq.empty[Row]) } } From 4c00a2b5386f318fbbded9e7f7e1daee7782758b Mon Sep 17 00:00:00 2001 From: "Jungtaek Lim (HeartSaVioR)" Date: Sat, 21 Sep 2019 08:26:20 +0900 Subject: [PATCH 5/5] Roll back unrelated change --- .../apache/spark/sql/hive/execution/AggregationQuerySuite.scala | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/sql/hive/src/test/scala/org/apache/spark/sql/hive/execution/AggregationQuerySuite.scala b/sql/hive/src/test/scala/org/apache/spark/sql/hive/execution/AggregationQuerySuite.scala index 86e1aeca9f6d6..4a3277f5a7e49 100644 --- a/sql/hive/src/test/scala/org/apache/spark/sql/hive/execution/AggregationQuerySuite.scala +++ b/sql/hive/src/test/scala/org/apache/spark/sql/hive/execution/AggregationQuerySuite.scala @@ -1021,7 +1021,7 @@ abstract class AggregationQuerySuite extends QueryTest with SQLTestUtils with Te test("SPARK-29122: hash-based aggregates for unfixed-length decimals in the interpreter mode") { withSQLConf(SQLConf.WHOLESTAGE_CODEGEN_ENABLED.key -> "false", - SQLConf.CODEGEN_FACTORY_MODE.key -> CodegenObjectFactoryMode.NO_CODEGEN.toString) { + SQLConf.CODEGEN_FACTORY_MODE.key -> CodegenObjectFactoryMode.NO_CODEGEN.toString) { withTempView("t") { spark.range(3).selectExpr("CAST(id AS decimal(38, 0)) a").createOrReplaceTempView("t") checkAnswer(sql("SELECT SUM(a) FROM t"), Row(java.math.BigDecimal.valueOf(3)))