Skip to content

[SPARK-29140][SQL] Handle parameters having "array" of javaType properly in splitAggregateExpressions #25830

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
HeartSaVioR wants to merge 5 commits into from
Closed

[SPARK-29140][SQL] Handle parameters having "array" of javaType properly in splitAggregateExpressions #25830

Show file tree
Hide file tree
Changes from 4 commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
de7249a
[SPARK-29140][SQL] Handle BinaryType of parameter properly in HashAgg…
HeartSaVioR Sep 18, 2019
9ce98b2
Reflect review comments
HeartSaVioR Sep 18, 2019
f045457
Additional review comment for nits
HeartSaVioR Sep 18, 2019
28726da
Reflect review comments
HeartSaVioR Sep 20, 2019
4c00a2b
Roll back unrelated change
HeartSaVioR Sep 20, 2019
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
8 changes: 8 additions & 0 deletions ...lyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/codegen/CodeGenerator.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
4 changes: 3 additions & 1 deletion sql/core/src/main/scala/org/apache/spark/sql/execution/aggregate/HashAggregateExec.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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"${CodeGenerator.typeName(v.javaType)} ${v.variableName}"
}.mkString(", ")
val doAggFuncName = ctx.addNewFunction(doAggFunc,
s"""
|private void $doAggFunc($argList) throws java.io.IOException {
Expand Down
17 changes: 16 additions & 1 deletion sql/hive/src/test/scala/org/apache/spark/sql/hive/execution/AggregationQuerySuite.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1021,13 +1021,28 @@ 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) {
Copy link
Member

@viirya viirya Sep 20, 2019

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Actually, I think previous is correct...

Copy link
Contributor Author

@HeartSaVioR HeartSaVioR Sep 20, 2019

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Yeah IDE automatically indented while fixing conflicts. Will roll back.

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 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, schema)
.groupBy($"id" % 10 as "group")
.agg(countDistinct($"c1"))
checkAnswer(aggDf, Seq.empty[Row])
}
}
}


Expand Down