Vivek1106-04 commented on code in PR #57528:
URL: https://github.com/apache/spark/pull/57528#discussion_r3654177960
##########
core/src/main/scala/org/apache/spark/util/Utils.scala:
##########
@@ -3152,6 +3152,59 @@ private[spark] object Utils
}
}
+ /**
+ * Return the n-th smallest element (0-indexed) of a long array, reordering
`sizes` in place.
+ *
+ * This is a quickselect, which runs in O(sizes.length) on average. Callers
that only need a few
+ * order statistics should prefer it over sorting the whole array, and must
not rely on the
+ * element order of `sizes` afterwards.
+ */
+ def nthSmallest(sizes: Array[Long], n: Int): Long = {
+ require(n >= 0 && n < sizes.length, s"n must be in [0, ${sizes.length})
but was $n")
+ var low = 0
+ var high = sizes.length - 1
+ while (low < high) {
+ // The middle element keeps already sorted and reverse sorted inputs,
which are both common
+ // for shuffle block sizes, away from the quadratic worst case.
+ val pivot = sizes(low + (high - low) / 2)
Review Comment:
Confirmed, and thanks for the organ pipe input specifically. Fixed in
2ced43b.
`nthSmallest` is now an introselect: median-of-three pivot, and it sorts the
range still under consideration once the ranges have stopped shrinking for `2 *
log2(n)` rounds. That bounds the worst case at O(n log n) regardless of the
input, rather than relying on the pivot choice being lucky.
One thing I want to be upfront about: I did not reproduce your comparison
counts or instrument the post-fix ones. What I added is a time bound on the
input you named -- six selections over a 2^20 organ pipe under an explicit 60
second limit, which currently run in 100ms total. Extrapolating your table
(50,455,932 at 16384, roughly quadrupling per doubling) puts a quadratic
selection at more than 10^11 comparisons there, so the limit is generous enough
not to be flaky while still failing decisively on a regression. If you would
rather have an actual comparison count asserted against a ceiling, I can add
one, though it means an instrumented copy of the loop in test code.
I also added a test asserting that `nthSmallest` leaves the array
partitioned around the returned element, since `HighlyCompressedMapStatus` now
counts the sizes above the cutoff by scanning only the tail of the array and so
depends on that ordering, not just on the returned value.
The benchmark now reports the organ pipe distribution alongside the skewed
one. At 50000 partitions it is 530us per map task against 310us for the
ordinary case -- a 1.7x gap, not a blowup.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
