Skip to content

[SQL][SPARK-17439] Fixing compression issues with approximate quantiles and adding more tests #15002

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
thunterdb wants to merge 3 commits into from
Closed

[SQL][SPARK-17439] Fixing compression issues with approximate quantiles and adding more tests #15002

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
75cb088
Actually call compress() in QuantileSummaries, and avoid expensive Ar…
srowen Sep 6, 2016
86afd44
work
thunterdb Sep 7, 2016
406943c
merging with @srowen's PR
thunterdb Sep 7, 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
16 changes: 12 additions & 4 deletions sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/util/QuantileSummaries.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -17,7 +17,7 @@

package org.apache.spark.sql.catalyst.util

import scala.collection.mutable.ArrayBuffer
import scala.collection.mutable.{ArrayBuffer, ListBuffer}

import org.apache.spark.sql.catalyst.util.QuantileSummaries.Stats

Expand Down Expand Up @@ -61,7 +61,12 @@ class QuantileSummaries(
def insert(x: Double): QuantileSummaries = {
headSampled += x
if (headSampled.size >= defaultHeadSize) {
this.withHeadBufferInserted
val result = this.withHeadBufferInserted
if (result.sampled.length >= compressThreshold) {
result.compress()
} else {
result
}
} else {
this
}
Expand Down Expand Up @@ -236,7 +241,7 @@ object QuantileSummaries {
if (currentSamples.isEmpty) {
return Array.empty[Stats]
}
val res: ArrayBuffer[Stats] = ArrayBuffer.empty
val res = ListBuffer.empty[Stats]
// Start for the last element, which is always part of the set.
// The head contains the current new head, that may be merged with the current element.
var head = currentSamples.last
Expand All @@ -258,7 +263,10 @@ object QuantileSummaries {
}
res.prepend(head)
// If necessary, add the minimum element:
res.prepend(currentSamples.head)
val currHead = currentSamples.head
if (currHead.value < head.value) {
Copy link
Contributor Author

@thunterdb thunterdb Sep 7, 2016

Choose a reason for hiding this comment

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

this is the fix

res.prepend(currentSamples.head)
}
res.toArray
}
}
29 changes: 27 additions & 2 deletions sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/util/QuantileSummariesSuite.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -40,6 +40,20 @@ class QuantileSummariesSuite extends SparkFunSuite {
summary.compress()
}

/**
* Interleaves compression and insertions.
*/
private def buildCompressSummary(
data: Seq[Double],
epsi: Double,
threshold: Int): QuantileSummaries = {
var summary = new QuantileSummaries(threshold, epsi)
data.foreach { x =>
summary = summary.insert(x).compress()
}
summary
}

private def checkQuantile(quant: Double, data: Seq[Double], summary: QuantileSummaries): Unit = {
val approx = summary.query(quant)
// The rank of the approximation.
Expand All @@ -54,8 +68,8 @@ class QuantileSummariesSuite extends SparkFunSuite {

for {
(seq_name, data) <- Seq(increasing, decreasing, random)
epsi <- Seq(0.1, 0.0001)
compression <- Seq(1000, 10)
epsi <- Seq(0.1, 0.0001) // With a significant value and with full precision
compression <- Seq(1000, 10) // This interleaves n so that we test without and with compression
} {

test(s"Extremas with epsi=$epsi and seq=$seq_name, compression=$compression") {
Expand All @@ -75,6 +89,17 @@ class QuantileSummariesSuite extends SparkFunSuite {
checkQuantile(0.1, data, s)
checkQuantile(0.001, data, s)
}

test(s"Some quantile values with epsi=$epsi and seq=$seq_name, compression=$compression " +
s"(interleaved)") {
val s = buildCompressSummary(data, epsi, compression)
assert(s.count == data.size, s"Found count=${s.count} but data size=${data.size}")
checkQuantile(0.9999, data, s)
checkQuantile(0.9, data, s)
checkQuantile(0.5, data, s)
checkQuantile(0.1, data, s)
checkQuantile(0.001, data, s)
}
}

// Tests for merging procedure
Expand Down