diff --git a/core/benchmarks/HighlyCompressedMapStatusBenchmark-jdk21-results.txt b/core/benchmarks/HighlyCompressedMapStatusBenchmark-jdk21-results.txt new file mode 100644 index 0000000000000..1abee6490442a --- /dev/null +++ b/core/benchmarks/HighlyCompressedMapStatusBenchmark-jdk21-results.txt @@ -0,0 +1,52 @@ +================================================================================================ +Build HighlyCompressedMapStatus, skewed block sizes +================================================================================================ + +OpenJDK 64-Bit Server VM 21.0.10 on Mac OS X 26.5.2 +Apple M4 +2048 shuffle partitions: Best Time(ms) Avg Time(ms) Stdev(ms) Rate(M/s) Per Row(ns) Relative +------------------------------------------------------------------------------------------------------------------------ +skewed block sizes not recorded 0 0 0 0.2 6458.0 1.0X +skewed block sizes recorded accurately 0 0 0 0.1 17250.0 0.4X + +OpenJDK 64-Bit Server VM 21.0.10 on Mac OS X 26.5.2 +Apple M4 +10000 shuffle partitions: Best Time(ms) Avg Time(ms) Stdev(ms) Rate(M/s) Per Row(ns) Relative +------------------------------------------------------------------------------------------------------------------------ +skewed block sizes not recorded 0 0 0 0.1 13834.0 1.0X +skewed block sizes recorded accurately 0 0 0 0.0 56250.0 0.2X + +OpenJDK 64-Bit Server VM 21.0.10 on Mac OS X 26.5.2 +Apple M4 +50000 shuffle partitions: Best Time(ms) Avg Time(ms) Stdev(ms) Rate(M/s) Per Row(ns) Relative +------------------------------------------------------------------------------------------------------------------------ +skewed block sizes not recorded 0 0 0 0.0 35959.0 1.0X +skewed block sizes recorded accurately 0 0 0 0.0 309750.0 0.1X + + +================================================================================================ +Build HighlyCompressedMapStatus, organ pipe block sizes +================================================================================================ + +OpenJDK 64-Bit Server VM 21.0.10 on Mac OS X 26.5.2 +Apple M4 +2048 shuffle partitions: Best Time(ms) Avg Time(ms) Stdev(ms) Rate(M/s) Per Row(ns) Relative +------------------------------------------------------------------------------------------------------------------------ +skewed block sizes not recorded 0 0 0 0.2 6416.0 1.0X +skewed block sizes recorded accurately 0 0 0 0.0 23583.0 0.3X + +OpenJDK 64-Bit Server VM 21.0.10 on Mac OS X 26.5.2 +Apple M4 +10000 shuffle partitions: Best Time(ms) Avg Time(ms) Stdev(ms) Rate(M/s) Per Row(ns) Relative +------------------------------------------------------------------------------------------------------------------------ +skewed block sizes not recorded 0 0 0 0.1 14666.0 1.0X +skewed block sizes recorded accurately 0 0 0 0.0 99458.0 0.1X + +OpenJDK 64-Bit Server VM 21.0.10 on Mac OS X 26.5.2 +Apple M4 +50000 shuffle partitions: Best Time(ms) Avg Time(ms) Stdev(ms) Rate(M/s) Per Row(ns) Relative +------------------------------------------------------------------------------------------------------------------------ +skewed block sizes not recorded 0 0 0 0.0 54583.0 1.0X +skewed block sizes recorded accurately 1 1 0 0.0 530209.0 0.1X + + diff --git a/core/src/main/scala/org/apache/spark/internal/config/package.scala b/core/src/main/scala/org/apache/spark/internal/config/package.scala index c8bd6410be27b..f18862880c8f3 100644 --- a/core/src/main/scala/org/apache/spark/internal/config/package.scala +++ b/core/src/main/scala/org/apache/spark/internal/config/package.scala @@ -1441,12 +1441,16 @@ package object config { ConfigBuilder("spark.shuffle.accurateBlockSkewedFactor") .doc("A shuffle block is considered as skewed and will be accurately recorded in " + "HighlyCompressedMapStatus if its size is larger than this factor multiplying " + - "the median shuffle block size or SHUFFLE_ACCURATE_BLOCK_THRESHOLD. It is " + - "recommended to set this parameter to be the same as SKEW_JOIN_SKEWED_PARTITION_FACTOR." + - "Set to -1.0 to disable this feature by default.") + "the median shuffle block size or SHUFFLE_ACCURATE_BLOCK_THRESHOLD. Otherwise every " + + "block below SHUFFLE_ACCURATE_BLOCK_THRESHOLD is reported as the average block size, " + + "which hides skew from adaptive query execution once a shuffle has more than " + + "SHUFFLE_MIN_NUM_PARTS_TO_HIGHLY_COMPRESS partitions. At most " + + "SHUFFLE_MAX_ACCURATE_SKEWED_BLOCK_NUMBER block sizes are recorded accurately per map " + + "task. The default matches SKEW_JOIN_SKEWED_PARTITION_FACTOR. Set to -1.0 to disable " + + "this feature.") .version("3.3.0") .doubleConf - .createWithDefault(-1.0) + .createWithDefault(5.0) private[spark] val SHUFFLE_MAX_ACCURATE_SKEWED_BLOCK_NUMBER = ConfigBuilder("spark.shuffle.maxAccurateSkewedBlockNumber") diff --git a/core/src/main/scala/org/apache/spark/scheduler/MapStatus.scala b/core/src/main/scala/org/apache/spark/scheduler/MapStatus.scala index e348b6a5f1493..3c27408334be0 100644 --- a/core/src/main/scala/org/apache/spark/scheduler/MapStatus.scala +++ b/core/src/main/scala/org/apache/spark/scheduler/MapStatus.scala @@ -286,10 +286,13 @@ private[spark] object HighlyCompressedMapStatus { Option(SparkEnv.get) .map(_.conf.get(config.SHUFFLE_ACCURATE_BLOCK_THRESHOLD)) .getOrElse(config.SHUFFLE_ACCURATE_BLOCK_THRESHOLD.defaultValue.get) + // Sizes at or above `threshold` are recorded accurately. Blocks of exactly `skewCutoff` are + // only recorded while `numTiedSkewedBlocksToRecord` lasts, because there can be arbitrarily + // many of them and recording them all would blow past SHUFFLE_MAX_ACCURATE_SKEWED_BLOCK_NUMBER. + var skewCutoff = Long.MaxValue + var numTiedSkewedBlocksToRecord = 0 val threshold = if (accurateBlockSkewedFactor > 0) { - val sortedSizes = uncompressedSizes.sorted - val medianSize: Long = Utils.median(sortedSizes, true) val maxAccurateSkewedBlockNumber = Math.min( Option(SparkEnv.get) @@ -297,11 +300,24 @@ private[spark] object HighlyCompressedMapStatus { .getOrElse(config.SHUFFLE_MAX_ACCURATE_SKEWED_BLOCK_NUMBER.defaultValue.get), totalNumBlocks ) + // Only two order statistics are needed here, so they are selected in O(totalNumBlocks) + // instead of sorting the sizes, which every map task would otherwise pay for. + val sizes = uncompressedSizes.clone() + val medianSize: Long = Utils.medianInPlace(sizes) + val firstAccurateIdx = totalNumBlocks - maxAccurateSkewedBlockNumber + skewCutoff = Utils.nthSmallest(sizes, firstAccurateIdx) + // At most `maxAccurateSkewedBlockNumber` sizes are strictly larger than `skewCutoff`, and + // they all sit above `firstAccurateIdx` now, so counting them bounds how many blocks of + // exactly `skewCutoff` may still be recorded. + var numLargerSizes = 0 + var j = firstAccurateIdx + while (j < totalNumBlocks) { + if (sizes(j) > skewCutoff) numLargerSizes += 1 + j += 1 + } + numTiedSkewedBlocksToRecord = maxAccurateSkewedBlockNumber - numLargerSizes val skewSizeThreshold = - Math.max( - medianSize * accurateBlockSkewedFactor, - sortedSizes(totalNumBlocks - maxAccurateSkewedBlockNumber).toDouble - ) + Math.max(medianSize * accurateBlockSkewedFactor, skewCutoff.toDouble) Math.min(shuffleAccurateBlockThreshold.toDouble, skewSizeThreshold) } else { // Disable skew detection if accurateBlockSkewedFactor <= 0 @@ -315,11 +331,21 @@ private[spark] object HighlyCompressedMapStatus { numNonEmptyBlocks += 1 // Huge blocks are not included in the calculation for average size, thus size for smaller // blocks is more accurate. - if (size < threshold) { + var isAccurate = size >= shuffleAccurateBlockThreshold + if (!isAccurate && size >= threshold) { + if (size > skewCutoff) { + isAccurate = true + } else if (numTiedSkewedBlocksToRecord > 0) { + // Ties are broken by block index so that the map status stays deterministic. + numTiedSkewedBlocksToRecord -= 1 + isAccurate = true + } + } + if (isAccurate) { + hugeBlockSizes(i) = MapStatus.compressSize(size) + } else { totalSmallBlockSize += size numSmallBlocks += 1 - } else { - hugeBlockSizes(i) = MapStatus.compressSize(uncompressedSizes(i)) } } else { emptyBlocks.add(i) diff --git a/core/src/main/scala/org/apache/spark/util/Utils.scala b/core/src/main/scala/org/apache/spark/util/Utils.scala index c8030e5a4c7c8..a2452b6424f97 100644 --- a/core/src/main/scala/org/apache/spark/util/Utils.scala +++ b/core/src/main/scala/org/apache/spark/util/Utils.scala @@ -3152,6 +3152,85 @@ private[spark] object Utils } } + /** + * Return the n-th smallest element (0-indexed) of a long array, reordering `sizes` in place. + * + * This is an introselect: a quickselect that gives up and sorts the range still under + * consideration once the ranges have stopped shrinking quickly enough, so it runs in + * O(sizes.length) on average and in O(sizes.length * log(sizes.length)) in the worst case. + * Callers that only need a few order statistics should prefer it over sorting the whole array. + * They must not rely on the element order of `sizes` afterwards, other than that no element + * before index n is larger and no element after it is smaller than the returned value. + */ + 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 + // A well chosen pivot roughly halves the range, so a good run takes about log2(sizes.length) + // rounds. Inputs such as an organ pipe distribution defeat any fixed pivot choice and would + // make plain quickselect quadratic, so stop partitioning after twice that many rounds. + var remainingRounds = 2 * (32 - Integer.numberOfLeadingZeros(sizes.length)) + while (low < high) { + if (remainingRounds == 0) { + java.util.Arrays.sort(sizes, low, high + 1) + low = n + high = n + } else { + remainingRounds -= 1 + // The median of the first, middle and last element keeps already sorted and reverse + // sorted inputs, which are both common for shuffle block sizes, on the linear path. + val pivot = medianOfThree(sizes, low, low + (high - low) / 2, high) + var i = low + var j = high + while (i <= j) { + while (sizes(i) < pivot) i += 1 + while (sizes(j) > pivot) j -= 1 + if (i <= j) { + val tmp = sizes(i) + sizes(i) = sizes(j) + sizes(j) = tmp + i += 1 + j -= 1 + } + } + if (n <= j) { + high = j + } else if (n >= i) { + low = i + } else { + // The n-th element sits between the two partitions, so it is in its final position. + low = n + high = n + } + } + } + sizes(n) + } + + /** Return the median of the elements of `sizes` at the three given positions. */ + private def medianOfThree(sizes: Array[Long], i: Int, j: Int, k: Int): Long = { + val a = sizes(i) + val b = sizes(j) + val c = sizes(k) + if (a < b) { + if (b < c) b else if (a < c) c else a + } else { + if (a < c) a else if (b < c) c else b + } + } + + /** + * Same as `median(sizes, false)`, but reorders `sizes` in place instead of sorting a copy of it. + */ + def medianInPlace(sizes: Array[Long]): Long = { + val len = sizes.length + len match { + case _ if (len % 2 == 0) => + math.max((nthSmallest(sizes, len / 2) + nthSmallest(sizes, len / 2 - 1)) / 2, 1) + case _ => math.max(nthSmallest(sizes, len / 2), 1) + } + } + /** * Check if a command is available. */ diff --git a/core/src/test/scala/org/apache/spark/HighlyCompressedMapStatusBenchmark.scala b/core/src/test/scala/org/apache/spark/HighlyCompressedMapStatusBenchmark.scala new file mode 100644 index 0000000000000..7257682a4d6c2 --- /dev/null +++ b/core/src/test/scala/org/apache/spark/HighlyCompressedMapStatusBenchmark.scala @@ -0,0 +1,94 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.spark + +import org.mockito.Mockito.{doReturn, mock} + +import org.apache.spark.benchmark.{Benchmark, BenchmarkBase} +import org.apache.spark.internal.config +import org.apache.spark.scheduler.HighlyCompressedMapStatus +import org.apache.spark.storage.BlockManagerId + +/** + * Benchmark for building a HighlyCompressedMapStatus, which every map task of a shuffle with more + * than spark.shuffle.minNumPartitionsToHighlyCompress partitions has to pay for. It covers both + * the case where skewed block sizes are recorded accurately and the case where they are not. + * {{{ + * To run this benchmark: + * 1. without sbt: bin/spark-submit --class + * 2. build/sbt "core/Test/runMain " + * 3. generate result: SPARK_GENERATE_BENCHMARK_FILES=1 build/sbt "core/Test/runMain " + * Results will be written to "benchmarks/HighlyCompressedMapStatusBenchmark-results.txt". + * }}} + */ +object HighlyCompressedMapStatusBenchmark extends BenchmarkBase { + + private val loc = BlockManagerId("a", "host", 1000) + + private def envWithSkewedFactor(skewedFactor: Double): SparkEnv = { + val conf = new SparkConf() + .set(config.SHUFFLE_ACCURATE_BLOCK_SKEWED_FACTOR.key, skewedFactor.toString) + val env = mock(classOf[SparkEnv]) + doReturn(conf, Seq.empty: _*).when(env).conf + env + } + + private def skewedBlockSizes(numPartitions: Int): Array[Long] = { + val r = new scala.util.Random(912) + Array.tabulate(numPartitions) { i => + // A few blocks are orders of magnitude larger than the rest, as in a skewed shuffle. + if (i % 500 == 0) (r.nextDouble() * 512 * 1024 * 1024).toLong + else (r.nextDouble() * 64 * 1024).toLong + } + } + + /** + * Sizes that rise and then fall. This is the worst case for selecting an order statistic with a + * fixed pivot choice, so it covers the cost of the fallback to sorting. + */ + private def organPipeBlockSizes(numPartitions: Int): Array[Long] = { + Array.tabulate(numPartitions)(i => Math.min(i, numPartitions - 1 - i).toLong) + } + + override def runBenchmarkSuite(mainArgs: Array[String]): Unit = { + val envWithoutSkewedSizes = envWithSkewedFactor(-1.0) + val envWithSkewedSizes = envWithSkewedFactor(5.0) + Seq("skewed" -> skewedBlockSizes _, "organ pipe" -> organPipeBlockSizes _).foreach { + case (distribution, blockSizes) => + runBenchmark(s"Build HighlyCompressedMapStatus, $distribution block sizes") { + Seq(2048, 10000, 50000).foreach { numPartitions => + val benchmark = new Benchmark(s"$numPartitions shuffle partitions", 1, output = output) + val sizes = blockSizes(numPartitions) + + benchmark.addCase("skewed block sizes not recorded", 10) { _ => + SparkEnv.set(envWithoutSkewedSizes) + HighlyCompressedMapStatus(loc, sizes, 0) + } + + benchmark.addCase("skewed block sizes recorded accurately", 10) { _ => + SparkEnv.set(envWithSkewedSizes) + HighlyCompressedMapStatus(loc, sizes, 0) + } + + benchmark.run() + } + } + } + SparkEnv.set(null) + } +} diff --git a/core/src/test/scala/org/apache/spark/scheduler/MapStatusSuite.scala b/core/src/test/scala/org/apache/spark/scheduler/MapStatusSuite.scala index 951c2c2d6cbe8..76915e4b832f8 100644 --- a/core/src/test/scala/org/apache/spark/scheduler/MapStatusSuite.scala +++ b/core/src/test/scala/org/apache/spark/scheduler/MapStatusSuite.scala @@ -289,4 +289,94 @@ class MapStatusSuite extends SparkFunSuite { "Only tracked skewed block size is accurate") } } + + test("SPARK-48290: skewed blocks are recorded accurately with the default configuration") { + val emptyBlocksLength = 3 + val smallBlocksLength = 3000 + val skewedBlocksLength = 5 + // Well above the median block size, but below SHUFFLE_ACCURATE_BLOCK_THRESHOLD, which is the + // case for a skewed partition whose rows are spread over a large number of map tasks. + val skewedBlockSize = 50 * 1024 * 1024L + + // No skew related config is set: this asserts the out of the box behavior. + val conf = new SparkConf() + val env = mock(classOf[SparkEnv]) + doReturn(conf).when(env).conf + SparkEnv.set(env) + + val emptyBlocks = createArray(emptyBlocksLength, 0L) + val smallBlocks = Array.tabulate[Long](smallBlocksLength)(i => i + 1) + val skewedBlocks = createArray(skewedBlocksLength, skewedBlockSize) + val allBlocks = emptyBlocks ++: smallBlocks ++: skewedBlocks + assert(skewedBlockSize < conf.get(config.SHUFFLE_ACCURATE_BLOCK_THRESHOLD), + "the skewed blocks must not be tracked as huge blocks") + val avg = smallBlocks.sum / smallBlocks.length + + val loc = BlockManagerId("a", "b", 10) + val mapTaskAttemptId = 5 + val status = compressAndDecompressMapStatus(MapStatus(loc, allBlocks, mapTaskAttemptId)) + assert(status.isInstanceOf[HighlyCompressedMapStatus]) + for (i <- 0 until emptyBlocksLength) { + assert(status.getSizeForBlock(i) === 0L) + } + for (i <- 0 until smallBlocksLength) { + assert(status.getSizeForBlock(emptyBlocksLength + i) === avg, + "the average size must not be inflated by the skewed blocks") + } + for (i <- 0 until skewedBlocksLength) { + assert(status.getSizeForBlock(emptyBlocksLength + smallBlocksLength + i) === + compressAndDecompressSize(skewedBlockSize), + "skewed block sizes must be accurate so that AQE can detect the skew") + } + } + + test("SPARK-48290: blocks tied at the cutoff size do not bypass the accurate skewed block " + + "limit") { + // The small blocks are the majority, so the median block size is smallBlockSize. Without the + // limit, all of the tied blocks would be recorded, which for a stage of 10000 map tasks would + // add more than a gigabyte of map status entries. + val smallBlocksLength = 25001 + val tiedBlocksLength = 24999 + val smallBlockSize = 1024L + // Far more blocks share this size than may be recorded, and it is the cutoff size itself: + // it is above the median times the skew factor, so the skew threshold is exactly this size. + val tiedBlockSize = 8 * 1024L + val maxAccurateSkewedBlockNumber = + config.SHUFFLE_MAX_ACCURATE_SKEWED_BLOCK_NUMBER.defaultValue.get + + // No skew related config is set: this asserts the out of the box behavior. + val conf = new SparkConf() + val env = mock(classOf[SparkEnv]) + doReturn(conf).when(env).conf + SparkEnv.set(env) + + val allBlocks = createArray(smallBlocksLength, smallBlockSize) ++: + createArray(tiedBlocksLength, tiedBlockSize) + assert(tiedBlockSize < conf.get(config.SHUFFLE_ACCURATE_BLOCK_THRESHOLD), + "the tied blocks must not be recorded as huge blocks") + assert(Utils.median(allBlocks, false) * + conf.get(config.SHUFFLE_ACCURATE_BLOCK_SKEWED_FACTOR) < tiedBlockSize, + "the cutoff size, not the median times the skew factor, must set the skew threshold") + val numSmallBlocks = allBlocks.length - maxAccurateSkewedBlockNumber + val avg = + (smallBlockSize * smallBlocksLength + + tiedBlockSize * (tiedBlocksLength - maxAccurateSkewedBlockNumber)) / numSmallBlocks + + val loc = BlockManagerId("a", "b", 10) + val status = compressAndDecompressMapStatus(MapStatus(loc, allBlocks, 5)) + assert(status.isInstanceOf[HighlyCompressedMapStatus]) + // Ties are broken by block index, so the first maxAccurateSkewedBlockNumber tied blocks are + // the ones recorded accurately. + val firstAccurateBlock = smallBlocksLength + for (i <- 0 until firstAccurateBlock) { + assert(status.getSizeForBlock(i) === avg) + } + for (i <- firstAccurateBlock until firstAccurateBlock + maxAccurateSkewedBlockNumber) { + assert(status.getSizeForBlock(i) === compressAndDecompressSize(tiedBlockSize)) + } + for (i <- firstAccurateBlock + maxAccurateSkewedBlockNumber until allBlocks.length) { + assert(status.getSizeForBlock(i) === avg, + "no more than maxAccurateSkewedBlockNumber blocks may be recorded accurately") + } + } } diff --git a/core/src/test/scala/org/apache/spark/util/UtilsSuite.scala b/core/src/test/scala/org/apache/spark/util/UtilsSuite.scala index e1e49a226735c..38778fb036503 100644 --- a/core/src/test/scala/org/apache/spark/util/UtilsSuite.scala +++ b/core/src/test/scala/org/apache/spark/util/UtilsSuite.scala @@ -39,6 +39,7 @@ import org.apache.hadoop.fs.audit.CommonAuditContext.currentAuditContext import org.apache.hadoop.ipc.{CallerContext => HadoopCallerContext} import org.apache.logging.log4j.Level import org.mockito.Mockito.when +import org.scalatest.time.SpanSugar._ import org.scalatestplus.mockito.MockitoSugar.mock import org.apache.spark.{SparkConf, SparkException, SparkFunSuite, TaskContext} @@ -1892,6 +1893,79 @@ class UtilsSuite extends SparkFunSuite with ResetSystemProperties { assert(new String(Files.readAllBytes(new File(dir, "file2.txt").toPath), UTF_8) === "world") } } + + test("SPARK-48290: nthSmallest returns the same elements as sorting") { + val inputs = Seq( + Array(1L), + Array(2L, 1L), + Array(5L, 4L, 3L, 2L, 1L), + Array(1L, 1L, 1L, 1L), + Array.tabulate(1000)(i => i.toLong), + Array.tabulate(1000)(i => (1000 - i).toLong), + Array.tabulate(1000)(_ => 7L), + Array.tabulate(1000)(i => math.min(i, 999 - i).toLong), + Array.fill(1000)(Random.nextLong())) + inputs.foreach { input => + val expected = input.sorted + for (n <- input.indices) { + assert(Utils.nthSmallest(input.clone(), n) === expected(n)) + } + } + } + + test("SPARK-48290: nthSmallest is not quadratic on an organ pipe distribution") { + // Sizes that rise and then fall defeat any fixed pivot choice, so a plain quickselect needs + // more than 10^11 comparisons for these selections and does not finish in any reasonable + // time. Sorting the range still under consideration keeps each selection in the millisecond + // range, so the time limit only has to be generous enough to be free of flakiness. + val length = 1 << 20 + val input = Array.tabulate(length)(i => math.min(i, length - 1 - i).toLong) + val expected = input.sorted + failAfter(60.seconds) { + Seq(0, 1, length / 2 - 1, length / 2, length - 100, length - 1).foreach { n => + assert(Utils.nthSmallest(input.clone(), n) === expected(n)) + } + } + } + + test("SPARK-48290: nthSmallest partitions the array around the returned element") { + // HighlyCompressedMapStatus counts the block sizes above the cutoff by scanning only the tail + // of the array, so it relies on this ordering and not just on the returned value. + val inputs = Seq( + Array.tabulate(1000)(i => i.toLong), + Array.tabulate(1000)(i => (1000 - i).toLong), + Array.tabulate(1000)(i => math.min(i, 999 - i).toLong), + Array.tabulate(1000)(i => (i % 7).toLong), + Array.fill(1000)(Random.nextLong())) + inputs.foreach { input => + Seq(0, 1, 17, 500, 900, 998, 999).foreach { n => + val sizes = input.clone() + val nth = Utils.nthSmallest(sizes, n) + assert(sizes(n) === nth) + assert(sizes.take(n).forall(_ <= nth), "no element below index n may be larger") + assert(sizes.drop(n + 1).forall(_ >= nth), "no element above index n may be smaller") + } + } + } + + test("SPARK-48290: nthSmallest rejects an out of range index") { + intercept[IllegalArgumentException](Utils.nthSmallest(Array(1L, 2L), -1)) + intercept[IllegalArgumentException](Utils.nthSmallest(Array(1L, 2L), 2)) + intercept[IllegalArgumentException](Utils.nthSmallest(Array.empty[Long], 0)) + } + + test("SPARK-48290: medianInPlace agrees with median") { + val inputs = Seq( + Array(1L), + Array(2L, 1L), + Array(0L, 0L, 0L), + Array(5L, 4L, 3L, 2L, 1L), + Array.tabulate(999)(i => i.toLong), + Array.fill(1000)(Random.nextLong().abs)) + inputs.foreach { input => + assert(Utils.medianInPlace(input.clone()) === Utils.median(input, false)) + } + } } private class SimpleExtension diff --git a/docs/configuration.md b/docs/configuration.md index 4999b08908dec..cad4b16e2db7f 100644 --- a/docs/configuration.md +++ b/docs/configuration.md @@ -1304,14 +1304,18 @@ Apart from these, the following properties are also available, and may be useful spark.shuffle.accurateBlockSkewedFactor - -1.0 + 5.0 A shuffle block is considered as skewed and will be accurately recorded in HighlyCompressedMapStatus if its size is larger than this factor multiplying - the median shuffle block size or spark.shuffle.accurateBlockThreshold. It is - recommended to set this parameter to be the same as - spark.sql.adaptive.skewJoin.skewedPartitionFactor. Set to -1.0 to disable this - feature by default. + the median shuffle block size or spark.shuffle.accurateBlockThreshold. + Otherwise every block below spark.shuffle.accurateBlockThreshold is reported + as the average block size, which hides skew from adaptive query execution once a shuffle + has more than spark.shuffle.minNumPartitionsToHighlyCompress partitions. At + most spark.shuffle.maxAccurateSkewedBlockNumber block sizes are recorded + accurately per map task. The default matches + spark.sql.adaptive.skewJoin.skewedPartitionFactor. Set to -1.0 to disable + this feature. 3.3.0