[SPARK-22285] [SQL] Change implementation of ApproxCountDistinctForIntervals to TypedImperativeAggregate #19506
Conversation
|
cc @cloud-fan |
|
test this please |
|
Test build #82798 has finished for PR 19506 at commit
|
|
Test build #82796 has finished for PR 19506 at commit
|
| override def prettyName: String = "approx_count_distinct_for_intervals" | ||
|
|
||
| override def serialize(obj: Array[Long]): Array[Byte] = { | ||
| val buffer = ByteBuffer.wrap(new Array(obj.length * Longs.BYTES)) |
There was a problem hiding this comment.
IIRC ByteBuffer is pretty slow for writing, shall we use unsafe writing?
There was a problem hiding this comment.
Changed to unsafe writing, could you take another look?
| val offset = mutableAggBufferOffset + hllppIndex * numWordsPerHllpp | ||
| hllppArray(hllppIndex).update(buffer, offset, value, child.dataType) | ||
| val offset = hllppIndex * numWordsPerHllpp | ||
| hllppArray(hllppIndex).update(LongArrayInput(buffer), offset, value, child.dataType) |
There was a problem hiding this comment.
you can just pass InternalRow(buffer) here, to save a lot of code changes. If performance matters here, you can create a LongArrayInternalRow to avoid boxing.
There was a problem hiding this comment.
InternalRow(buffer) will copy the buffer.
Creating a LongArrayInternalRow is a good idea, thanks!
This reverts commit ba75112.
|
Test build #82929 has finished for PR 19506 at commit
|
|
Test build #82947 has finished for PR 19506 at commit
|
|
Test build #82948 has finished for PR 19506 at commit
|
|
|
||
| override def serialize(obj: Array[Long]): Array[Byte] = { | ||
| val byteArray = new Array[Byte](obj.length * 8) | ||
| obj.indices.foreach { i => |
There was a problem hiding this comment.
use while loop here for better performance in Scala, as this is a performance sensitive code path.
There was a problem hiding this comment.
Fixed. Thanks for the reminder!
| override def deserialize(bytes: Array[Byte]): Array[Long] = { | ||
| val length = bytes.length / 8 | ||
| val longArray = new Array[Long](length) | ||
| (0 until length).foreach { i => |
| } | ||
|
|
||
| override def deserialize(bytes: Array[Byte]): Array[Long] = { | ||
| val length = bytes.length / 8 |
There was a problem hiding this comment.
add assert(bytes.length % 8 == 0)
|
LGTM except a few minor comments |
|
Test build #82966 has finished for PR 19506 at commit
|
|
thanks, merging to master! |
3 participants
What changes were proposed in this pull request?
The current implementation of
ApproxCountDistinctForIntervalsisImperativeAggregate. The number ofaggBufferAttributesis the number of total words in the hllppHelper array. Each hllppHelper has 52 words by default relativeSD.Since this aggregate function is used in equi-height histogram generation, and the number of buckets in histogram is usually hundreds, the number of
aggBufferAttributescan easily reach tens of thousands or even more.This leads to a huge method in codegen and causes error:
Besides, huge generated methods also result in performance regression.
In this PR, we change its implementation to
TypedImperativeAggregate. After the fix,ApproxCountDistinctForIntervalscan deal with more than thousands endpoints without throwing codegen error, and improve performance from20 secto2 secin a test case of 500 endpoints.How was this patch tested?
Test by an added test case and existing tests.