Skip to content

[SPARK-16283][SQL] Implements percentile_approx aggregation function which supports partial aggregation. #14868

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 6 commits into from
Closed

[SPARK-16283][SQL] Implements percentile_approx aggregation function which supports partial aggregation. #14868

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
6 commits
Select commit Hold shift + click to select a range
6be2ebe
percentile approximate
clockfly Aug 19, 2016
3dda1ae
fix UT
clockfly Aug 30, 2016
96bd5a4
address comments
clockfly Aug 30, 2016
539d693
address comments
clockfly Aug 30, 2016
1965b1a
address comments
clockfly Aug 30, 2016
3f08c02
address wenchen's comment
clockfly Aug 31, 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 UT
  • Loading branch information
@clockfly
clockfly committed Aug 30, 2016
commit 3dda1ae90cea0eab8dd963d4af5708bff8846d65
18 changes: 10 additions & 8 deletions ...ain/scala/org/apache/spark/sql/catalyst/expressions/aggregate/ApproximatePercentile.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -91,8 +91,10 @@ case class ApproximatePercentile(
// Rule ImplicitTypeCasts can cast other numeric types to double
case (_, num: Double) => (false, Array(num))
case (ArrayType(baseType: NumericType, _), arrayData: ArrayData) =>
val numericArray = arrayData.toArray(baseType)(baseType.classTag)
(true, numericArray.map(baseType.numeric.toDouble))
val numericArray = arrayData.toObjectArray(baseType)
(true, numericArray.map {x =>
baseType.numeric.toDouble(x.asInstanceOf[baseType.InternalType])
})
case other =>
throw new AnalysisException(s"Invalid data type ${other._1} for parameter percentage")
}
Expand Down Expand Up @@ -183,8 +185,8 @@ object ApproximatePercentile {
* underlying quantileSummaries is compressed.
*/
class PercentileDigest(
private var summaries: QuantileSummaries,
private var isCompressed: Boolean) {
private var summaries: QuantileSummaries,
private var isCompressed: Boolean) {
Copy link
Contributor

@cloud-fan cloud-fan Sep 1, 2016

Choose a reason for hiding this comment

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

why we need this flag? QuantileSummaries provides a way to show it's compressed or not: headSampled.isEmpty

Copy link
Contributor Author

@clockfly clockfly Sep 1, 2016

Choose a reason for hiding this comment

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

  1. headSampled is private.
  2. headSample.isEmpty not necessary means it is compressed. (Some bug in QuantileSummaries)


// Trigger compression if the QuantileSummaries's buffer length exceeds
// compressThresHoldBufferLength. The buffer length can be get by
Expand Down Expand Up @@ -287,10 +289,10 @@ object ApproximatePercentile {
// An case class to wrap fields of QuantileSummaries, so that we can use the expression encoder
// to serialize it.
case class QuantileSummariesData(
Copy link
Contributor Author

@clockfly clockfly Aug 29, 2016 •
edited
Loading

Choose a reason for hiding this comment

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

QuantileSummaries contains some private var fields, so I think it is better to define a wrapper class instead of changing QuantileSummaries to case class (case class is not supposed to contain var fields).

val compressThreshold: Int,
val relativeError: Double,
val sampled: Array[Stats] = Array.empty,
val count: Long = 0L) {
compressThreshold: Int,
relativeError: Double,
sampled: Array[Stats],
count: Long) {
def this(summary: QuantileSummaries) = {
this(summary.compressThreshold, summary.relativeError, summary.sampled, summary.count)
}
Expand Down
6 changes: 5 additions & 1 deletion sql/hive/src/test/scala/org/apache/spark/sql/catalyst/ExpressionToSQLSuite.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -155,7 +155,11 @@ class ExpressionToSQLSuite extends SQLBuilderTest with SQLTestUtils {

test("aggregate functions") {
checkSqlGeneration("SELECT approx_count_distinct(value) FROM t1 GROUP BY key")
checkSqlGeneration("SELECT percentile_approx(value) FROM t1 GROUP BY key")
checkSqlGeneration("SELECT percentile_approx(value, 0.25) FROM t1 GROUP BY key")
checkSqlGeneration("SELECT percentile_approx(value, array(0.25, 0.75)) FROM t1 GROUP BY key")
checkSqlGeneration("SELECT percentile_approx(value, 0.25, 100) FROM t1 GROUP BY key")
checkSqlGeneration(
"SELECT percentile_approx(value, array(0.25, 0.75), 100) FROM t1 GROUP BY key")
checkSqlGeneration("SELECT avg(value) FROM t1 GROUP BY key")
checkSqlGeneration("SELECT corr(value, key) FROM t1 GROUP BY key")
checkSqlGeneration("SELECT count(value) FROM t1 GROUP BY key")
Expand Down