Skip to content

[SPARK-27653][SQL] Add max_by() and min_by() SQL aggregate functions #24557

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
viirya wants to merge 6 commits into from
Closed

[SPARK-27653][SQL] Add max_by() and min_by() SQL aggregate functions #24557

Changes from 1 commit
Commits
Show all changes
6 commits
Select commit Hold shift + click to select a range
5c7e3c5
Add max_by.
viirya May 8, 2019
79f1015
Add min_by.
viirya May 9, 2019
798f0fa
Add test.
viirya May 9, 2019
26b5a32
Using DSL.
viirya May 11, 2019
dd1d9de
Address comments.
viirya May 13, 2019
05f1767
Renaming variables.
viirya May 13, 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
Prev Previous commit
Renaming variables.
  • Loading branch information
@viirya
viirya committed May 13, 2019
commit 05f1767dc49e6caebd91abf207014bcd8029b4f1
41 changes: 21 additions & 20 deletions ...st/src/main/scala/org/apache/spark/sql/catalyst/expressions/aggregate/MaxByAndMinBy.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -49,45 +49,46 @@ abstract class MaxMinBy extends DeclarativeAggregate {
TypeUtils.checkForOrderingExpr(orderingExpr.dataType, s"function $funcName")

// The attributes used to keep extremum (max or min) and associated aggregated values.
Copy link
Contributor

@cloud-fan cloud-fan May 13, 2019

Choose a reason for hiding this comment

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

ah that's a good point. Shall we call it extremumOrdering then?

Copy link
Member Author

@viirya viirya May 13, 2019

Choose a reason for hiding this comment

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

Good for me. +1

private lazy val maxOrdering = AttributeReference("maxOrdering", orderingExpr.dataType)()
private lazy val valueWithMaxOrdering =
AttributeReference("valueWithMaxOrdering", valueExpr.dataType)()
private lazy val extremumOrdering =
AttributeReference("extremumOrdering", orderingExpr.dataType)()
private lazy val valueWithExtremumOrdering =
AttributeReference("valueWithExtremumOrdering", valueExpr.dataType)()

override lazy val aggBufferAttributes: Seq[AttributeReference] =
valueWithMaxOrdering :: maxOrdering :: Nil
valueWithExtremumOrdering :: extremumOrdering :: Nil

private lazy val nullValue = Literal.create(null, valueExpr.dataType)
private lazy val nullOrdering = Literal.create(null, orderingExpr.dataType)

override lazy val initialValues: Seq[Literal] = Seq(
/* valueWithMaxOrdering = */ nullValue,
/* maxOrdering = */ nullOrdering
/* valueWithExtremumOrdering = */ nullValue,
/* extremumOrdering = */ nullOrdering
)

override lazy val updateExpressions: Seq[Expression] = Seq(
/* valueWithMaxOrdering = */
/* valueWithExtremumOrdering = */
CaseWhen(
(maxOrdering.isNull && orderingExpr.isNull, nullValue) ::
(maxOrdering.isNull, valueExpr) ::
(orderingExpr.isNull, valueWithMaxOrdering) :: Nil,
If(predicate(maxOrdering, orderingExpr), valueWithMaxOrdering, valueExpr)
(extremumOrdering.isNull && orderingExpr.isNull, nullValue) ::
(extremumOrdering.isNull, valueExpr) ::
(orderingExpr.isNull, valueWithExtremumOrdering) :: Nil,
If(predicate(extremumOrdering, orderingExpr), valueWithExtremumOrdering, valueExpr)
),
/* maxOrdering = */ orderingUpdater(maxOrdering, orderingExpr)
/* extremumOrdering = */ orderingUpdater(extremumOrdering, orderingExpr)
)

override lazy val mergeExpressions: Seq[Expression] = Seq(
/* valueWithMaxOrdering = */
/* valueWithExtremumOrdering = */
CaseWhen(
(maxOrdering.left.isNull && maxOrdering.right.isNull, nullValue) ::
(maxOrdering.left.isNull, valueWithMaxOrdering.right) ::
(maxOrdering.right.isNull, valueWithMaxOrdering.left) :: Nil,
If(predicate(maxOrdering.left, maxOrdering.right),
valueWithMaxOrdering.left, valueWithMaxOrdering.right)
(extremumOrdering.left.isNull && extremumOrdering.right.isNull, nullValue) ::
(extremumOrdering.left.isNull, valueWithExtremumOrdering.right) ::
(extremumOrdering.right.isNull, valueWithExtremumOrdering.left) :: Nil,
If(predicate(extremumOrdering.left, extremumOrdering.right),
valueWithExtremumOrdering.left, valueWithExtremumOrdering.right)
),
/* maxOrdering = */ orderingUpdater(maxOrdering.left, maxOrdering.right)
/* extremumOrdering = */ orderingUpdater(extremumOrdering.left, extremumOrdering.right)
)

override lazy val evaluateExpression: AttributeReference = valueWithMaxOrdering
override lazy val evaluateExpression: AttributeReference = valueWithExtremumOrdering
}

@ExpressionDescription(
Expand Down