Add a downward-only shuffle partition pass sized from consumer-stage shuffle input

core/src/main/resources/bootstrap/tuningConfigs.yaml

@@ -218,6 +218,35 @@ default:
     default: 2
     usedBy: spark.sql.shuffle.partitions
 
+  # Downward shuffle-partition tuning configs.
+  # These drive the final downward-only pass that lowers an oversized shuffle partition
+  # recommendation based on the total uncompressed shuffle input of the worst consumer stage.
+  - name: DOWNWARD_SHUFFLE_ENABLED
+    description: >-
+      Enables the final downward-only shuffle partition pass. When disabled, the AutoTuner
+      never lowers the shuffle partition recommendation produced by the normal tuning passes.
+      Disabled by default, so enabling it is a per-run opt-in. What has to be settled before it
+      defaults to true is the target partition size, not a count of validated applications: on a
+      large workload the hand-tuned value, the ColumnarExchange bound and this pass at a 1 GiB
+      target disagree, and this pass asks for the most partitions of the three.
+    default: false
+    usedBy: spark.sql.shuffle.partitions, spark.sql.adaptive.coalescePartitions.initialPartitionNum
+
+  - name: DOWNWARD_SHUFFLE_TARGET_PARTITION_SIZE
+    description: >-
+      Target amount of estimated GPU shuffle input processed by a single partition. The
+      per-stage partition requirement is the estimated stage input divided by this value.
+    default: 1g
+    usedBy: spark.sql.shuffle.partitions, spark.sql.adaptive.coalescePartitions.initialPartitionNum
+
+  - name: DOWNWARD_SHUFFLE_INPUT_SIZE_FACTOR
+    description: >-
+      Factor applied to the measured uncompressed shuffle input to estimate the GPU input size.
+      Profiling reads measured GpuColumnarExchange data, so it uses 1.0. Qualification reads CPU
+      Exchange data and overrides this in the qualification section.
+    default: 1.0
+    usedBy: spark.sql.shuffle.partitions, spark.sql.adaptive.coalescePartitions.initialPartitionNum
+
   - name: WORKER_GPU_COUNT
     description: >-
       Default number of GPUs per worker node
@@ -450,6 +479,13 @@ qualification:
     default: 1g
     usedBy: spark.rapids.sql.batchSizeBytes
 
+  - name: DOWNWARD_SHUFFLE_INPUT_SIZE_FACTOR
+    description: >-
+      Qualification reads uncompressed CPU Exchange data, so the GPU input size is estimated
+      rather than measured. This factor is an initial conservative estimate.
+    default: 0.8
+    usedBy: spark.sql.shuffle.partitions, spark.sql.adaptive.coalescePartitions.initialPartitionNum
+
 # Profiling tool specific tuning configs
 profiling:
   - name: BATCH_SIZE_BYTES

core/src/main/scala/com/nvidia/spark/rapids/tool/AppSummaryInfoBaseProvider.scala

@@ -17,10 +17,12 @@
 package com.nvidia.spark.rapids.tool
 
 import com.nvidia.spark.rapids.tool.analysis.AggRawMetricsResult
+import com.nvidia.spark.rapids.tool.analysis.AppSQLPlanAnalyzer
 import com.nvidia.spark.rapids.tool.profiling.{AppInfoColumnarExchangeMetrics,
   AppInfoJobStageAggMetricsVisitor, AppInfoPropertyGetter, AppInfoReadMetrics,
-  AppInfoSqlTaskAggMetricsVisitor, AppInfoSQLTaskInputSizes, BaseProfilingAppSummaryInfoProvider,
-  DataSourceProfileResult, ProfilerResult, PySparkMemoryEvidence, SingleAppSummaryInfoProvider}
+  AppInfoShuffleStageInputMetrics, AppInfoSqlTaskAggMetricsVisitor, AppInfoSQLTaskInputSizes,
+  BaseProfilingAppSummaryInfoProvider, DataSourceProfileResult, ProfilerResult,
+  PySparkMemoryEvidence, SingleAppSummaryInfoProvider}
 import com.nvidia.spark.rapids.tool.tuning.QualAppSummaryInfoProvider
 
 import org.apache.spark.sql.rapids.tool.ToolUtils
@@ -35,7 +37,8 @@ class AppSummaryInfoBaseProvider extends AppInfoPropertyGetter
   with AppInfoSqlTaskAggMetricsVisitor
   with AppInfoSQLTaskInputSizes
   with AppInfoReadMetrics
-  with AppInfoColumnarExchangeMetrics {
+  with AppInfoColumnarExchangeMetrics
+  with AppInfoShuffleStageInputMetrics {
   def isAppInfoAvailable = false
   override def getAllProperties: Map[String, String] = Map[String, String]()
   override def getSparkProperty(propKey: String): Option[String] = None
@@ -89,12 +92,15 @@ object AppSummaryInfoBaseProvider {
    * tool.
    * @param appInfo
    * @param appAggStats optional aggregate of application stats
+   * @param sqlAnalyzer the SQL plan analyzer already built for this application, reused so the
+   *                    plans are not traversed a second time
    * @return object that can be used by the AutoTuner to calculate the recommendations
    */
   def fromQualAppInfo(appInfo: QualificationAppInfo,
       appAggStats: Option[QualificationSummaryInfo] = None,
       rawAggMetrics: AggRawMetricsResult,
-      dsInfo: Seq[DataSourceProfileResult]): AppSummaryInfoBaseProvider = {
-    new QualAppSummaryInfoProvider(appInfo, appAggStats, rawAggMetrics, dsInfo)
+      dsInfo: Seq[DataSourceProfileResult],
+      sqlAnalyzer: Option[AppSQLPlanAnalyzer] = None): AppSummaryInfoBaseProvider = {
+    new QualAppSummaryInfoProvider(appInfo, appAggStats, rawAggMetrics, dsInfo, sqlAnalyzer)
   }
 }

core/src/main/scala/com/nvidia/spark/rapids/tool/analysis/AppSQLPlanAnalyzer.scala

@@ -1,5 +1,5 @@
 /*
- * Copyright (c) 2024-2025, NVIDIA CORPORATION.
+ * Copyright (c) 2024-2026, NVIDIA CORPORATION.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -20,7 +20,7 @@ import scala.collection.mutable.{AbstractSet, ArrayBuffer, HashMap, LinkedHashSe
 
 import com.nvidia.spark.rapids.tool.analysis.util.IOAccumDiagnosticMetrics._
 import com.nvidia.spark.rapids.tool.analysis.util.StageAccumDiagnosticMetrics._
-import com.nvidia.spark.rapids.tool.profiling.{AccumProfileResults, IODiagnosticResult, SQLAccumProfileResults, SQLMetricInfoCase, SQLStageInfoProfileResult, UnsupportedSQLPlan, WholeStageCodeGenResults}
+import com.nvidia.spark.rapids.tool.profiling.{AccumProfileResults, IODiagnosticResult, ShuffleStageInputAnalysis, SQLAccumProfileResults, SQLMetricInfoCase, SQLStageInfoProfileResult, UnsupportedSQLPlan, WholeStageCodeGenResults}
 
 import org.apache.spark.sql.rapids.tool.{AppBase, RDDCheckHelper}
 import org.apache.spark.sql.rapids.tool.plangraph.{SparkPlanGraphCluster, SparkPlanGraphNode, ToolsPlanGraph}
@@ -68,6 +68,13 @@ class AppSQLPlanAnalyzer(app: AppBase)
   val IODiagnosticMetricsMap: HashMap[(Long, Long), ArrayBuffer[SQLAccumProfileResults]] =
     HashMap.empty[(Long, Long), ArrayBuffer[SQLAccumProfileResults]]
 
+  /**
+   * Raw consumer-stage shuffle input inventory used by the AutoTuner's downward shuffle-partition
+   * pass. It is computed lazily and cached so that both tools reuse this analyzer's plan graphs
+   * instead of traversing the SQL plans a second time.
+   */
+  lazy val shuffleStageInputAnalysis: ShuffleStageInputAnalysis = ShuffleStageInputAnalyzer(app)
+
   /**
    * Updates the stageToDiagnosticMetrics mapping with the provided AccumProfileResults.
    * @param accum AccumProfileResults instance containing diagnostic metrics to be added