Skip to content

[SPARK-17187][SQL] Supports using arbitrary Java object as internal aggregation buffer object #14753

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
clockfly wants to merge 12 commits into from
Closed

[SPARK-17187][SQL] Supports using arbitrary Java object as internal aggregation buffer object #14753

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
12 commits
Select commit Hold shift + click to select a range
10861b2
object aggregation buffer
clockfly Aug 19, 2016
0fdc1ea
fix comments
clockfly Aug 23, 2016
d3108ab
fix review comments
clockfly Aug 23, 2016
2873765
fix review comments
clockfly Aug 23, 2016
7190eb0
fix review comments
clockfly Aug 24, 2016
5904bcd
on viirya's comment
clockfly Aug 24, 2016
8c8bd9a
on yin's comment
clockfly Aug 24, 2016
7e7cb85
On wenchen's comment
clockfly Aug 24, 2016
86166a1
On wenchen's comment
clockfly Aug 24, 2016
e060d21
On wenchen's comment
clockfly Aug 24, 2016
ac8e36a
add test for nullable aggregation function
clockfly Aug 25, 2016
ca574e1
use while loop
yhuai Aug 25, 2016
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
Prev Previous commit
Next Next commit
fix review comments
  • Loading branch information
@clockfly
clockfly committed Aug 24, 2016
commit 7190eb0c2a4dce2c5b84c29fb90bb2def23a3520
9 changes: 8 additions & 1 deletion ...alyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/aggregate/interfaces.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -432,6 +432,12 @@ abstract class DeclarativeAggregate
* 4. After processing all input aggregation objects of current group (group by key), the framework
* calls method `eval(buffer: T)` to generate the final output for this group.
* 5. The framework moves on to next group, until all groups have been processed.
*
* NOTE: SQL with TypedImperativeAggregate functions is planned in sort based aggregation,
* instead of hash based aggregation, as TypedImperativeAggregate use BinaryType as aggregation
* buffer's storage format, which is not supported by hash based aggregation. Hash based
* aggregation only support aggregation buffer of mutable types (like LongType, IntType that have
* fixed length and can be mutated in place in UnsafeRow)
*/
abstract class TypedImperativeAggregate[T] extends ImperativeAggregate {
Copy link
Contributor

@hvanhovell hvanhovell Aug 25, 2016

Choose a reason for hiding this comment

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

Isn't this the wrong way around? Isn't ImperativeAggregate the untyped version of an TypedImperativeAggregate? Much like Dataset and DataFrame?

I know this has been done for engineering purposes, but I still wonder if we shouldn't reverse the hierarchy here.

Copy link
Contributor

@yhuai yhuai Aug 25, 2016

Choose a reason for hiding this comment

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

ImperativeAggregate only defines the interface. It does not specify what are accepted buffer types, right?


Expand Down Expand Up @@ -507,8 +513,9 @@ abstract class TypedImperativeAggregate[T] extends ImperativeAggregate {
}
}

private[this] val anyObjectType = ObjectType(classOf[AnyRef])
private def getField[U](input: InternalRow, fieldIndex: Int): U = {
Copy link
Contributor

@yhuai yhuai Aug 24, 2016

Choose a reason for hiding this comment

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

Seems we only need getField(input: InternalRow, fieldIndex: Int): T?

input.get(fieldIndex, null).asInstanceOf[U]
input.get(fieldIndex, anyObjectType).asInstanceOf[U]
}

final override lazy val aggBufferAttributes: Seq[AttributeReference] = {
Expand Down
11 changes: 2 additions & 9 deletions sql/core/src/main/scala/org/apache/spark/sql/execution/aggregate/AggUtils.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -54,16 +54,9 @@ object AggUtils {
initialInputBufferOffset: Int = 0,
resultExpressions: Seq[NamedExpression] = Nil,
child: SparkPlan): SparkPlan = {

val hasTypedImperativeAggregate: Boolean = aggregateExpressions.exists {
case AggregateExpression(agg: TypedImperativeAggregate[_], _, _, _) => true
case _ => false
}

val aggBufferAttributesSupportedByHashAggregate = HashAggregateExec.supportsAggregate(
val useHash = HashAggregateExec.supportsAggregate(
aggregateExpressions.flatMap(_.aggregateFunction.aggBufferAttributes))

if (aggBufferAttributesSupportedByHashAggregate && !hasTypedImperativeAggregate) {
if (useHash) {
HashAggregateExec(
requiredChildDistributionExpressions = requiredChildDistributionExpressions,
groupingExpressions = groupingExpressions,
Expand Down
9 changes: 1 addition & 8 deletions ...rc/main/scala/org/apache/spark/sql/execution/aggregate/SortBasedAggregationIterator.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -55,14 +55,7 @@ class SortBasedAggregationIterator(

val genericMutableBuffer = new GenericMutableRow(bufferRowSize)

val allFieldsMutable = bufferSchema.map(_.dataType).forall(UnsafeRow.isMutable)

val hasTypedImperativeAggregate = aggregateFunctions.exists {
case agg: TypedImperativeAggregate[_] => true
case _ => false
}

val useUnsafeBuffer = allFieldsMutable && !hasTypedImperativeAggregate
val useUnsafeBuffer = bufferSchema.map(_.dataType).forall(UnsafeRow.isMutable)

val buffer = if (useUnsafeBuffer) {
val unsafeProjection =
Expand Down