This is an automated email from the ASF dual-hosted git repository.
pan3793 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/master by this push:
new ae2a97cbb83d [SPARK-58100][SQL] Use IndexedSeq to fix quadratic
complexity in partition resolution
ae2a97cbb83d is described below
commit ae2a97cbb83d89580e81ec21e36dead02589e500
Author: Daniil Filippov <[email protected]>
AuthorDate: Wed Jul 15 11:41:09 2026 +0800
[SPARK-58100][SQL] Use IndexedSeq to fix quadratic complexity in partition
resolution
This PR fixes the issue
[57220](https://github.com/apache/spark/issues/57220).
### What changes were proposed in this pull request?
The PR updates `resolveTypeConflicts` to return
`IndexedSeq[TypedPartValue]` (backed by `Vector`) instead of `Seq`.
`Vector.apply(index)` is effectively O(1), reducing the overall complexity to
O(nk).
### Why are the changes needed?
For now, `resolvedValues` is a `Seq[Seq[TypedPartValue]]`, where each inner
collection is a linked list. So the `apply` method called there is O(n). This
is called for every partition and every column, making the total complexity
O(n²k), where n = number of partitions and k = number of partition columns.
With thousands of daily partitions, this becomes a real issue: a thread
"indefinitely" spins at the following stack trace:
```
PartitioningUtils.resolvePartitions
- resolvedValues.map(_(index))
- List.apply(index)
- List.drop(index)
- StrictOptimizedLinearSeqOps.loop$2
```
### Does this PR introduce _any_ user-facing change?
No
### How was this patch tested?
The following benchmark confirms the issue and the fix:
```scala
package org.apache.spark.sql.execution.datasources
import org.apache.hadoop.fs.Path
import org.apache.spark.benchmark.{Benchmark, BenchmarkBase}
import
org.apache.spark.sql.execution.datasources.PartitioningUtils.{PartitionValues,
TypedPartValue}
import org.apache.spark.sql.types.IntegerType
/**
* Benchmark for [[PartitioningUtils.resolvePartitions]] to guard against
O(n²) regressions.
*
* See https://github.com/apache/spark/issues/57220 for details.
*
* To run this benchmark:
* {{{
* 1. build/sbt "sql/Test/runMain <this class>"
* 2. generate result: SPARK_GENERATE_BENCHMARK_FILES=1 build/sbt
"sql/Test/runMain <this class>"
* Results will be written to
"benchmarks/PartitioningUtilsResolvePartitionsBenchmark-results.txt".
* }}}
*/
object PartitioningUtilsResolvePartitionsBenchmark extends BenchmarkBase {
override def runBenchmarkSuite(mainArgs: Array[String]): Unit = {
val partitionCounts = Seq(5000, 10000, 15000)
resolvePartitionsBenchmark(partitionCounts, numCols = 1)
resolvePartitionsBenchmark(partitionCounts, numCols = 3)
}
private def resolvePartitionsBenchmark(partitionCounts: Seq[Int],
numCols: Int): Unit = {
val benchmarkName = s"resolvePartitions with $numCols partition
column(s)"
runBenchmark(benchmarkName) {
val benchmark = new Benchmark(benchmarkName, partitionCounts.last,
output = output)
partitionCounts.foreach { n =>
val input = buildInput(n, numCols)
benchmark.addCase(s"$n partitions") { _ =>
PartitioningUtils.resolvePartitions(input, caseSensitive = true)
}
}
benchmark.run()
}
}
private def buildInput(n: Int, numCols: Int): List[(Path,
PartitionValues)] = {
val colNames = List.range(0, numCols).map(c => s"col$c")
List.range(0, n).map { i =>
val typedValues = (0 until numCols).map(c => TypedPartValue(s"${i +
c}", IntegerType))
val partitionedPart = colNames
.zip(typedValues)
.map { case (k, v) => s"$k=${v.value}" }
.mkString("/")
val path = new
Path(s"file:/tmp/resolve_partitions_benchmark/$partitionedPart")
path -> PartitionValues(colNames, typedValues)
}
}
}
```
Results before the fix:
```
================================================================================================
resolvePartitions with 1 partition column(s)
================================================================================================
OpenJDK 64-Bit Server VM 21.0.11+10-LTS on Mac OS X 26.5.2
Apple M4 Pro
resolvePartitions with 1 partition column(s): Best Time(ms) Avg Time(ms)
Stdev(ms) Rate(M/s) Per Row(ns) Relative
----------------------------------------------------------------------------------------------------------------------------
5000 partitions 19 20
1 0.8 1283.7 1.0X
10000 partitions 79 81
2 0.2 5282.2 0.2X
15000 partitions 179 187
6 0.1 11952.9 0.1X
================================================================================================
resolvePartitions with 3 partition column(s)
================================================================================================
OpenJDK 64-Bit Server VM 21.0.11+10-LTS on Mac OS X 26.5.2
Apple M4 Pro
resolvePartitions with 3 partition column(s): Best Time(ms) Avg Time(ms)
Stdev(ms) Rate(M/s) Per Row(ns) Relative
----------------------------------------------------------------------------------------------------------------------------
5000 partitions 60 64
3 0.3 3994.3 1.0X
10000 partitions 254 258
3 0.1 16948.2 0.2X
15000 partitions 524 529
5 0.0 34925.0 0.1X
```
Results after the fix:
```
================================================================================================
resolvePartitions with 1 partition column(s)
================================================================================================
OpenJDK 64-Bit Server VM 21.0.11+10-LTS on Mac OS X 26.5.2
Apple M4 Pro
resolvePartitions with 1 partition column(s): Best Time(ms) Avg Time(ms)
Stdev(ms) Rate(M/s) Per Row(ns) Relative
----------------------------------------------------------------------------------------------------------------------------
5000 partitions 0 0
0 70.0 14.3 1.0X
10000 partitions 0 0
0 34.5 29.0 0.5X
15000 partitions 1 1
0 23.5 42.5 0.3X
================================================================================================
resolvePartitions with 3 partition column(s)
================================================================================================
OpenJDK 64-Bit Server VM 21.0.11+10-LTS on Mac OS X 26.5.2
Apple M4 Pro
resolvePartitions with 3 partition column(s): Best Time(ms) Avg Time(ms)
Stdev(ms) Rate(M/s) Per Row(ns) Relative
----------------------------------------------------------------------------------------------------------------------------
5000 partitions 0 1
0 34.2 29.3 1.0X
10000 partitions 1 1
0 16.0 62.5 0.5X
15000 partitions 1 2
0 10.1 99.4 0.3X
```
### Was this patch authored or co-authored using generative AI tooling?
No
Closes #57243 from flipp5b/fix-partition-resolution.
Lead-authored-by: Daniil Filippov <[email protected]>
Co-authored-by: Daniil Filippov <[email protected]>
Signed-off-by: Cheng Pan <[email protected]>
---
.../apache/spark/sql/execution/datasources/PartitioningUtils.scala | 5 +++--
1 file changed, 3 insertions(+), 2 deletions(-)
diff --git
a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/PartitioningUtils.scala
b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/PartitioningUtils.scala
index 83a094c250d6..53dac003754c 100644
---
a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/PartitioningUtils.scala
+++
b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/PartitioningUtils.scala
@@ -648,11 +648,12 @@ object PartitioningUtils extends SQLConfHelper {
* Given a collection of [[Literal]]s, resolves possible type conflicts by
* [[findWiderTypeForPartitionColumn]].
*/
- private def resolveTypeConflicts(typedValues: Seq[TypedPartValue]):
Seq[TypedPartValue] = {
+ private def resolveTypeConflicts(typedValues: Seq[TypedPartValue]):
IndexedSeq[TypedPartValue] = {
val dataTypes = typedValues.map(_.dataType)
val desiredType = dataTypes.reduce(findWiderTypeForPartitionColumn)
- typedValues.map(tv => tv.copy(dataType = desiredType))
+ // IndexedSeq guarantees O(1) apply at the call site; a List would make
the loop O(n^2).
+ typedValues.view.map(tv => tv.copy(dataType = desiredType)).toIndexedSeq
}
/**
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
