Hi all, I'm a new contributor, and I've opened a PR for SPARK-48290, which has been open since May 2024 with no PR attached:
JIRA: https://issues.apache.org/jira/browse/SPARK-48290 PR: https://github.com/apache/spark/pull/57528 The report is that AQE stops detecting skewed joins once a query uses more than 2000 shuffle partitions. I reproduced it on master and the cause turned out not to be in AQE at all. Above spark.shuffle.minNumPartitionsToHighlyCompress (2000), map statuses switch to HighlyCompressedMapStatus, which records exact sizes only for blocks above spark.shuffle.accurateBlockThreshold (100MB) and reports the average block size for everything else. A skewed partition spread over many map tasks has no single block above 100MB, so its size is averaged away. MapOutputTracker.getStatistics then sums those, and AQE sees min == median == max, which is exactly the log line in the report. This also explains why lowering accurateBlockThreshold doesn't help: what matters is the per-block size, not the per-partition total. Varying only spark.sql.shuffle.partitions, with a join where 95% of rows land on one key: 1999 partitions, defaults -> skew detected 2001 partitions, defaults -> not detected 4000 partitions, defaults -> not detected 2001, spark.shuffle.accurateBlockSkewedFactor=5.0 -> detected 2001, minNumPartitionsToHighlyCompress=100000 -> detected Either knob restores detection on its own, which points at the map status compression rather than at AQE. The mechanism to fix this already exists. SPARK-36967 added spark.shuffle.accurateBlockSkewedFactor in 3.3 to keep skewed block sizes accurate, but it ships disabled, even though its own documentation recommends setting it to the same value as spark.sql.adaptive.skewJoin.skewedPartitionFactor. Enabling it sorted the block sizes in every map task, which I assume is why it was left off. So the PR does two things: it replaces that sort with selection of the only two order statistics the code needs, which is O(numPartitions) instead of O(numPartitions log numPartitions), and then enables the feature by default. Per map task at 50000 partitions the accurate path goes from 1562us to 378us, and the driver carries at most ~500 bytes more per map status. Full numbers are in the PR along with a new benchmark. Since this changes a default and therefore query plans, it needs committer judgement rather than just a code review. I'd appreciate it if someone familiar with map output statistics or AQE skew handling could take a look. I'm happy to split the algorithmic change and the default flip into separate PRs if that's easier to review, or to add an AQE level test if the runtime cost is acceptable. Thanks, Vivek
