diff --git a/dev/diffs/3.4.3.diff b/dev/diffs/3.4.3.diff
index 8fc6451ef12..5da66b80bc8 100644
--- a/dev/diffs/3.4.3.diff
+++ b/dev/diffs/3.4.3.diff
@@ -1563,7 +1563,7 @@ index ac710c32296..2854b433dd3 100644
  
    import testImplicits._
 diff --git a/sql/core/src/test/scala/org/apache/spark/sql/execution/adaptive/AdaptiveQueryExecSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/execution/adaptive/AdaptiveQueryExecSuite.scala
-index 593bd7bb4ba..b327d84d5cc 100644
+index 593bd7bb4ba..9467f497cb9 100644
 --- a/sql/core/src/test/scala/org/apache/spark/sql/execution/adaptive/AdaptiveQueryExecSuite.scala
 +++ b/sql/core/src/test/scala/org/apache/spark/sql/execution/adaptive/AdaptiveQueryExecSuite.scala
 @@ -26,9 +26,11 @@ import org.scalatest.time.SpanSugar._
@@ -1674,17 +1674,7 @@ index 593bd7bb4ba..b327d84d5cc 100644
      withSQLConf(
        SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "true",
        SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "80",
-@@ -273,7 +289,8 @@ class AdaptiveQueryExecSuite
-       val localReads = collect(adaptivePlan) {
-         case read: AQEShuffleReadExec if read.isLocalRead => read
-       }
--      assert(localReads.length == 2)
-+      // Comet shuffle changes shuffle metrics
-+      assert(localReads.length == 1)
-       val localShuffleRDD0 = localReads(0).execute().asInstanceOf[ShuffledRowRDD]
-       val localShuffleRDD1 = localReads(1).execute().asInstanceOf[ShuffledRowRDD]
-       // the final parallelism is math.max(1, numReduces / numMappers): math.max(1, 5/2) = 2
-@@ -298,7 +315,9 @@ class AdaptiveQueryExecSuite
+@@ -298,7 +314,9 @@ class AdaptiveQueryExecSuite
            .groupBy($"a").count()
          checkAnswer(testDf, Seq())
          val plan = testDf.queryExecution.executedPlan
@@ -1695,7 +1685,7 @@ index 593bd7bb4ba..b327d84d5cc 100644
          val coalescedReads = collect(plan) {
            case r: AQEShuffleReadExec => r
          }
-@@ -312,7 +331,9 @@ class AdaptiveQueryExecSuite
+@@ -312,7 +330,9 @@ class AdaptiveQueryExecSuite
            .groupBy($"a").count()
          checkAnswer(testDf, Seq())
          val plan = testDf.queryExecution.executedPlan
@@ -1706,7 +1696,7 @@ index 593bd7bb4ba..b327d84d5cc 100644
          val coalescedReads = collect(plan) {
            case r: AQEShuffleReadExec => r
          }
-@@ -322,7 +343,7 @@ class AdaptiveQueryExecSuite
+@@ -322,7 +342,7 @@ class AdaptiveQueryExecSuite
      }
    }
  
@@ -1715,7 +1705,7 @@ index 593bd7bb4ba..b327d84d5cc 100644
      withSQLConf(
          SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "true",
          SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "80") {
-@@ -337,7 +358,7 @@ class AdaptiveQueryExecSuite
+@@ -337,7 +357,7 @@ class AdaptiveQueryExecSuite
      }
    }
  
@@ -1724,7 +1714,7 @@ index 593bd7bb4ba..b327d84d5cc 100644
      withSQLConf(
          SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "true",
          SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "80") {
-@@ -353,7 +374,7 @@ class AdaptiveQueryExecSuite
+@@ -353,7 +373,7 @@ class AdaptiveQueryExecSuite
      }
    }
  
@@ -1733,7 +1723,7 @@ index 593bd7bb4ba..b327d84d5cc 100644
      withSQLConf(
          SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "true",
          SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "80") {
-@@ -398,7 +419,7 @@ class AdaptiveQueryExecSuite
+@@ -398,7 +418,7 @@ class AdaptiveQueryExecSuite
      }
    }
  
@@ -1742,7 +1732,7 @@ index 593bd7bb4ba..b327d84d5cc 100644
      withSQLConf(
          SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "true",
          SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "80") {
-@@ -443,7 +464,7 @@ class AdaptiveQueryExecSuite
+@@ -443,7 +463,7 @@ class AdaptiveQueryExecSuite
      }
    }
  
@@ -1751,7 +1741,7 @@ index 593bd7bb4ba..b327d84d5cc 100644
      withSQLConf(
          SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "true",
          SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "500") {
-@@ -489,7 +510,7 @@ class AdaptiveQueryExecSuite
+@@ -489,7 +509,7 @@ class AdaptiveQueryExecSuite
      }
    }
  
@@ -1760,7 +1750,7 @@ index 593bd7bb4ba..b327d84d5cc 100644
      withSQLConf(
          SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "true",
          SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "80") {
-@@ -508,7 +529,7 @@ class AdaptiveQueryExecSuite
+@@ -508,7 +528,7 @@ class AdaptiveQueryExecSuite
      }
    }
  
@@ -1769,7 +1759,7 @@ index 593bd7bb4ba..b327d84d5cc 100644
      withSQLConf(
          SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "true",
          SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "80") {
-@@ -539,7 +560,9 @@ class AdaptiveQueryExecSuite
+@@ -539,7 +559,9 @@ class AdaptiveQueryExecSuite
        assert(smj.size == 1)
        val bhj = findTopLevelBroadcastHashJoin(adaptivePlan)
        assert(bhj.size == 1)
@@ -1780,7 +1770,7 @@ index 593bd7bb4ba..b327d84d5cc 100644
        // Even with local shuffle read, the query stage reuse can also work.
        val ex = findReusedExchange(adaptivePlan)
        assert(ex.nonEmpty)
-@@ -560,7 +583,9 @@ class AdaptiveQueryExecSuite
+@@ -560,7 +582,9 @@ class AdaptiveQueryExecSuite
        assert(smj.size == 1)
        val bhj = findTopLevelBroadcastHashJoin(adaptivePlan)
        assert(bhj.size == 1)
@@ -1791,7 +1781,7 @@ index 593bd7bb4ba..b327d84d5cc 100644
        // Even with local shuffle read, the query stage reuse can also work.
        val ex = findReusedExchange(adaptivePlan)
        assert(ex.isEmpty)
-@@ -569,7 +594,8 @@ class AdaptiveQueryExecSuite
+@@ -569,7 +593,8 @@ class AdaptiveQueryExecSuite
      }
    }
  
@@ -1801,7 +1791,7 @@ index 593bd7bb4ba..b327d84d5cc 100644
      withSQLConf(
          SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "true",
          SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "20000000",
-@@ -664,7 +690,8 @@ class AdaptiveQueryExecSuite
+@@ -664,7 +689,8 @@ class AdaptiveQueryExecSuite
        val bhj = findTopLevelBroadcastHashJoin(adaptivePlan)
        assert(bhj.size == 1)
        // There is still a SMJ, and its two shuffles can't apply local read.
@@ -1811,7 +1801,7 @@ index 593bd7bb4ba..b327d84d5cc 100644
      }
    }
  
-@@ -786,7 +813,8 @@ class AdaptiveQueryExecSuite
+@@ -786,7 +812,8 @@ class AdaptiveQueryExecSuite
      }
    }
  
@@ -1821,7 +1811,7 @@ index 593bd7bb4ba..b327d84d5cc 100644
      Seq("SHUFFLE_MERGE", "SHUFFLE_HASH").foreach { joinHint =>
        def getJoinNode(plan: SparkPlan): Seq[ShuffledJoin] = if (joinHint == "SHUFFLE_MERGE") {
          findTopLevelSortMergeJoin(plan)
-@@ -1004,7 +1032,8 @@ class AdaptiveQueryExecSuite
+@@ -1004,7 +1031,8 @@ class AdaptiveQueryExecSuite
      }
    }
  
@@ -1831,7 +1821,7 @@ index 593bd7bb4ba..b327d84d5cc 100644
      withSQLConf(SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "true") {
        val (_, adaptivePlan) = runAdaptiveAndVerifyResult(
          "SELECT key FROM testData GROUP BY key")
-@@ -1599,7 +1628,7 @@ class AdaptiveQueryExecSuite
+@@ -1599,7 +1627,7 @@ class AdaptiveQueryExecSuite
          val (_, adaptivePlan) = runAdaptiveAndVerifyResult(
            "SELECT id FROM v1 GROUP BY id DISTRIBUTE BY id")
          assert(collect(adaptivePlan) {
@@ -1840,7 +1830,7 @@ index 593bd7bb4ba..b327d84d5cc 100644
          }.length == 1)
        }
      }
-@@ -1679,7 +1708,8 @@ class AdaptiveQueryExecSuite
+@@ -1679,7 +1707,8 @@ class AdaptiveQueryExecSuite
      }
    }
  
@@ -1850,7 +1840,7 @@ index 593bd7bb4ba..b327d84d5cc 100644
      def hasRepartitionShuffle(plan: SparkPlan): Boolean = {
        find(plan) {
          case s: ShuffleExchangeLike =>
-@@ -1864,6 +1894,9 @@ class AdaptiveQueryExecSuite
+@@ -1864,6 +1893,9 @@ class AdaptiveQueryExecSuite
      def checkNoCoalescePartitions(ds: Dataset[Row], origin: ShuffleOrigin): Unit = {
        assert(collect(ds.queryExecution.executedPlan) {
          case s: ShuffleExchangeExec if s.shuffleOrigin == origin && s.numPartitions == 2 => s
@@ -1860,7 +1850,7 @@ index 593bd7bb4ba..b327d84d5cc 100644
        }.size == 1)
        ds.collect()
        val plan = ds.queryExecution.executedPlan
-@@ -1872,6 +1905,9 @@ class AdaptiveQueryExecSuite
+@@ -1872,6 +1904,9 @@ class AdaptiveQueryExecSuite
        }.isEmpty)
        assert(collect(plan) {
          case s: ShuffleExchangeExec if s.shuffleOrigin == origin && s.numPartitions == 2 => s
@@ -1870,7 +1860,7 @@ index 593bd7bb4ba..b327d84d5cc 100644
        }.size == 1)
        checkAnswer(ds, testData)
      }
-@@ -2028,7 +2064,8 @@ class AdaptiveQueryExecSuite
+@@ -2028,7 +2063,8 @@ class AdaptiveQueryExecSuite
      }
    }
  
@@ -1880,7 +1870,7 @@ index 593bd7bb4ba..b327d84d5cc 100644
      withTempView("t1", "t2") {
        def checkJoinStrategy(shouldShuffleHashJoin: Boolean): Unit = {
          Seq("100", "100000").foreach { size =>
-@@ -2114,7 +2151,8 @@ class AdaptiveQueryExecSuite
+@@ -2114,7 +2150,8 @@ class AdaptiveQueryExecSuite
      }
    }
  
@@ -1890,7 +1880,7 @@ index 593bd7bb4ba..b327d84d5cc 100644
      withTempView("v") {
        withSQLConf(
          SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "true",
-@@ -2213,7 +2251,7 @@ class AdaptiveQueryExecSuite
+@@ -2213,7 +2250,7 @@ class AdaptiveQueryExecSuite
                runAdaptiveAndVerifyResult(s"SELECT $repartition key1 FROM skewData1 " +
                  s"JOIN skewData2 ON key1 = key2 GROUP BY key1")
              val shuffles1 = collect(adaptive1) {
@@ -1899,7 +1889,7 @@ index 593bd7bb4ba..b327d84d5cc 100644
              }
              assert(shuffles1.size == 3)
              // shuffles1.head is the top-level shuffle under the Aggregate operator
-@@ -2226,7 +2264,7 @@ class AdaptiveQueryExecSuite
+@@ -2226,7 +2263,7 @@ class AdaptiveQueryExecSuite
                runAdaptiveAndVerifyResult(s"SELECT $repartition key1 FROM skewData1 " +
                  s"JOIN skewData2 ON key1 = key2")
              val shuffles2 = collect(adaptive2) {
@@ -1908,7 +1898,7 @@ index 593bd7bb4ba..b327d84d5cc 100644
              }
              if (hasRequiredDistribution) {
                assert(shuffles2.size == 3)
-@@ -2260,7 +2298,8 @@ class AdaptiveQueryExecSuite
+@@ -2260,7 +2297,8 @@ class AdaptiveQueryExecSuite
      }
    }
  
@@ -1918,7 +1908,7 @@ index 593bd7bb4ba..b327d84d5cc 100644
      CostEvaluator.instantiate(
        classOf[SimpleShuffleSortCostEvaluator].getCanonicalName, spark.sparkContext.getConf)
      intercept[IllegalArgumentException] {
-@@ -2404,6 +2443,7 @@ class AdaptiveQueryExecSuite
+@@ -2404,6 +2442,7 @@ class AdaptiveQueryExecSuite
            val (_, adaptive) = runAdaptiveAndVerifyResult(query)
            assert(adaptive.collect {
              case sort: SortExec => sort
@@ -1926,7 +1916,7 @@ index 593bd7bb4ba..b327d84d5cc 100644
            }.size == 1)
            val read = collect(adaptive) {
              case read: AQEShuffleReadExec => read
-@@ -2421,7 +2461,8 @@ class AdaptiveQueryExecSuite
+@@ -2421,7 +2460,8 @@ class AdaptiveQueryExecSuite
      }
    }
  
@@ -1936,7 +1926,7 @@ index 593bd7bb4ba..b327d84d5cc 100644
      withTempView("v") {
        withSQLConf(
          SQLConf.ADAPTIVE_OPTIMIZE_SKEWS_IN_REBALANCE_PARTITIONS_ENABLED.key -> "true",
-@@ -2533,7 +2574,7 @@ class AdaptiveQueryExecSuite
+@@ -2533,7 +2573,7 @@ class AdaptiveQueryExecSuite
            runAdaptiveAndVerifyResult("SELECT key1 FROM skewData1 JOIN skewData2 ON key1 = key2 " +
              "JOIN skewData3 ON value2 = value3")
          val shuffles1 = collect(adaptive1) {
@@ -1945,7 +1935,7 @@ index 593bd7bb4ba..b327d84d5cc 100644
          }
          assert(shuffles1.size == 4)
          val smj1 = findTopLevelSortMergeJoin(adaptive1)
-@@ -2544,7 +2585,7 @@ class AdaptiveQueryExecSuite
+@@ -2544,7 +2584,7 @@ class AdaptiveQueryExecSuite
            runAdaptiveAndVerifyResult("SELECT key1 FROM skewData1 JOIN skewData2 ON key1 = key2 " +
              "JOIN skewData3 ON value1 = value3")
          val shuffles2 = collect(adaptive2) {
@@ -1954,7 +1944,7 @@ index 593bd7bb4ba..b327d84d5cc 100644
          }
          assert(shuffles2.size == 4)
          val smj2 = findTopLevelSortMergeJoin(adaptive2)
-@@ -2703,7 +2744,8 @@ class AdaptiveQueryExecSuite
+@@ -2703,7 +2743,8 @@ class AdaptiveQueryExecSuite
      }
    }
  
diff --git a/dev/diffs/3.5.9.diff b/dev/diffs/3.5.9.diff
index 462554a1047..4e3c60e3748 100644
--- a/dev/diffs/3.5.9.diff
+++ b/dev/diffs/3.5.9.diff
@@ -1561,7 +1561,7 @@ index 5a413c77754..207b66e1d7b 100644
  
    import testImplicits._
 diff --git a/sql/core/src/test/scala/org/apache/spark/sql/execution/adaptive/AdaptiveQueryExecSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/execution/adaptive/AdaptiveQueryExecSuite.scala
-index 2f8e401e743..7849c685b19 100644
+index 2f8e401e743..29dc002bc80 100644
 --- a/sql/core/src/test/scala/org/apache/spark/sql/execution/adaptive/AdaptiveQueryExecSuite.scala
 +++ b/sql/core/src/test/scala/org/apache/spark/sql/execution/adaptive/AdaptiveQueryExecSuite.scala
 @@ -27,9 +27,11 @@ import org.scalatest.time.SpanSugar._
@@ -1679,17 +1679,7 @@ index 2f8e401e743..7849c685b19 100644
      withSQLConf(
        SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "true",
        SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "80",
-@@ -299,7 +316,8 @@ class AdaptiveQueryExecSuite
-       val localReads = collect(adaptivePlan) {
-         case read: AQEShuffleReadExec if read.isLocalRead => read
-       }
--      assert(localReads.length == 2)
-+      // Comet shuffle changes shuffle metrics
-+      assert(localReads.length == 1)
-       val localShuffleRDD0 = localReads(0).execute().asInstanceOf[ShuffledRowRDD]
-       val localShuffleRDD1 = localReads(1).execute().asInstanceOf[ShuffledRowRDD]
-       // the final parallelism is math.max(1, numReduces / numMappers): math.max(1, 5/2) = 2
-@@ -324,7 +342,9 @@ class AdaptiveQueryExecSuite
+@@ -324,7 +341,9 @@ class AdaptiveQueryExecSuite
            .groupBy($"a").count()
          checkAnswer(testDf, Seq())
          val plan = testDf.queryExecution.executedPlan
@@ -1700,7 +1690,7 @@ index 2f8e401e743..7849c685b19 100644
          val coalescedReads = collect(plan) {
            case r: AQEShuffleReadExec => r
          }
-@@ -338,7 +358,9 @@ class AdaptiveQueryExecSuite
+@@ -338,7 +357,9 @@ class AdaptiveQueryExecSuite
            .groupBy($"a").count()
          checkAnswer(testDf, Seq())
          val plan = testDf.queryExecution.executedPlan
@@ -1711,7 +1701,7 @@ index 2f8e401e743..7849c685b19 100644
          val coalescedReads = collect(plan) {
            case r: AQEShuffleReadExec => r
          }
-@@ -348,7 +370,7 @@ class AdaptiveQueryExecSuite
+@@ -348,7 +369,7 @@ class AdaptiveQueryExecSuite
      }
    }
  
@@ -1720,7 +1710,7 @@ index 2f8e401e743..7849c685b19 100644
      withSQLConf(
          SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "true",
          SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "80") {
-@@ -363,7 +385,7 @@ class AdaptiveQueryExecSuite
+@@ -363,7 +384,7 @@ class AdaptiveQueryExecSuite
      }
    }
  
@@ -1729,7 +1719,7 @@ index 2f8e401e743..7849c685b19 100644
      withSQLConf(
          SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "true",
          SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "80") {
-@@ -379,7 +401,7 @@ class AdaptiveQueryExecSuite
+@@ -379,7 +400,7 @@ class AdaptiveQueryExecSuite
      }
    }
  
@@ -1738,7 +1728,7 @@ index 2f8e401e743..7849c685b19 100644
      withSQLConf(
          SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "true",
          SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "80") {
-@@ -424,7 +446,7 @@ class AdaptiveQueryExecSuite
+@@ -424,7 +445,7 @@ class AdaptiveQueryExecSuite
      }
    }
  
@@ -1747,7 +1737,7 @@ index 2f8e401e743..7849c685b19 100644
      withSQLConf(
          SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "true",
          SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "80") {
-@@ -469,7 +491,7 @@ class AdaptiveQueryExecSuite
+@@ -469,7 +490,7 @@ class AdaptiveQueryExecSuite
      }
    }
  
@@ -1756,7 +1746,7 @@ index 2f8e401e743..7849c685b19 100644
      withSQLConf(
          SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "true",
          SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "500") {
-@@ -515,7 +537,7 @@ class AdaptiveQueryExecSuite
+@@ -515,7 +536,7 @@ class AdaptiveQueryExecSuite
      }
    }
  
@@ -1765,7 +1755,7 @@ index 2f8e401e743..7849c685b19 100644
      withSQLConf(
          SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "true",
          SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "80") {
-@@ -534,7 +556,7 @@ class AdaptiveQueryExecSuite
+@@ -534,7 +555,7 @@ class AdaptiveQueryExecSuite
      }
    }
  
@@ -1774,7 +1764,7 @@ index 2f8e401e743..7849c685b19 100644
      withSQLConf(
          SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "true",
          SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "80") {
-@@ -565,7 +587,9 @@ class AdaptiveQueryExecSuite
+@@ -565,7 +586,9 @@ class AdaptiveQueryExecSuite
        assert(smj.size == 1)
        val bhj = findTopLevelBroadcastHashJoin(adaptivePlan)
        assert(bhj.size == 1)
@@ -1785,7 +1775,7 @@ index 2f8e401e743..7849c685b19 100644
        // Even with local shuffle read, the query stage reuse can also work.
        val ex = findReusedExchange(adaptivePlan)
        assert(ex.nonEmpty)
-@@ -586,7 +610,9 @@ class AdaptiveQueryExecSuite
+@@ -586,7 +609,9 @@ class AdaptiveQueryExecSuite
        assert(smj.size == 1)
        val bhj = findTopLevelBroadcastHashJoin(adaptivePlan)
        assert(bhj.size == 1)
@@ -1796,7 +1786,7 @@ index 2f8e401e743..7849c685b19 100644
        // Even with local shuffle read, the query stage reuse can also work.
        val ex = findReusedExchange(adaptivePlan)
        assert(ex.isEmpty)
-@@ -595,7 +621,8 @@ class AdaptiveQueryExecSuite
+@@ -595,7 +620,8 @@ class AdaptiveQueryExecSuite
      }
    }
  
@@ -1806,7 +1796,7 @@ index 2f8e401e743..7849c685b19 100644
      withSQLConf(
          SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "true",
          SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "20000000",
-@@ -690,7 +717,8 @@ class AdaptiveQueryExecSuite
+@@ -690,7 +716,8 @@ class AdaptiveQueryExecSuite
        val bhj = findTopLevelBroadcastHashJoin(adaptivePlan)
        assert(bhj.size == 1)
        // There is still a SMJ, and its two shuffles can't apply local read.
@@ -1816,7 +1806,7 @@ index 2f8e401e743..7849c685b19 100644
      }
    }
  
-@@ -812,7 +840,8 @@ class AdaptiveQueryExecSuite
+@@ -812,7 +839,8 @@ class AdaptiveQueryExecSuite
      }
    }
  
@@ -1826,7 +1816,7 @@ index 2f8e401e743..7849c685b19 100644
      Seq("SHUFFLE_MERGE", "SHUFFLE_HASH").foreach { joinHint =>
        def getJoinNode(plan: SparkPlan): Seq[ShuffledJoin] = if (joinHint == "SHUFFLE_MERGE") {
          findTopLevelSortMergeJoin(plan)
-@@ -1030,7 +1059,8 @@ class AdaptiveQueryExecSuite
+@@ -1030,7 +1058,8 @@ class AdaptiveQueryExecSuite
      }
    }
  
@@ -1836,7 +1826,7 @@ index 2f8e401e743..7849c685b19 100644
      withSQLConf(SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "true") {
        val (_, adaptivePlan) = runAdaptiveAndVerifyResult(
          "SELECT key FROM testData GROUP BY key")
-@@ -1625,7 +1655,7 @@ class AdaptiveQueryExecSuite
+@@ -1625,7 +1654,7 @@ class AdaptiveQueryExecSuite
          val (_, adaptivePlan) = runAdaptiveAndVerifyResult(
            "SELECT id FROM v1 GROUP BY id DISTRIBUTE BY id")
          assert(collect(adaptivePlan) {
@@ -1845,7 +1835,7 @@ index 2f8e401e743..7849c685b19 100644
          }.length == 1)
        }
      }
-@@ -1705,7 +1735,8 @@ class AdaptiveQueryExecSuite
+@@ -1705,7 +1734,8 @@ class AdaptiveQueryExecSuite
      }
    }
  
@@ -1855,7 +1845,7 @@ index 2f8e401e743..7849c685b19 100644
      def hasRepartitionShuffle(plan: SparkPlan): Boolean = {
        find(plan) {
          case s: ShuffleExchangeLike =>
-@@ -1890,6 +1921,9 @@ class AdaptiveQueryExecSuite
+@@ -1890,6 +1920,9 @@ class AdaptiveQueryExecSuite
      def checkNoCoalescePartitions(ds: Dataset[Row], origin: ShuffleOrigin): Unit = {
        assert(collect(ds.queryExecution.executedPlan) {
          case s: ShuffleExchangeExec if s.shuffleOrigin == origin && s.numPartitions == 2 => s
@@ -1865,7 +1855,7 @@ index 2f8e401e743..7849c685b19 100644
        }.size == 1)
        ds.collect()
        val plan = ds.queryExecution.executedPlan
-@@ -1898,6 +1932,9 @@ class AdaptiveQueryExecSuite
+@@ -1898,6 +1931,9 @@ class AdaptiveQueryExecSuite
        }.isEmpty)
        assert(collect(plan) {
          case s: ShuffleExchangeExec if s.shuffleOrigin == origin && s.numPartitions == 2 => s
@@ -1875,7 +1865,7 @@ index 2f8e401e743..7849c685b19 100644
        }.size == 1)
        checkAnswer(ds, testData)
      }
-@@ -2054,7 +2091,8 @@ class AdaptiveQueryExecSuite
+@@ -2054,7 +2090,8 @@ class AdaptiveQueryExecSuite
      }
    }
  
@@ -1885,7 +1875,7 @@ index 2f8e401e743..7849c685b19 100644
      withTempView("t1", "t2") {
        def checkJoinStrategy(shouldShuffleHashJoin: Boolean): Unit = {
          Seq("100", "100000").foreach { size =>
-@@ -2140,7 +2178,8 @@ class AdaptiveQueryExecSuite
+@@ -2140,7 +2177,8 @@ class AdaptiveQueryExecSuite
      }
    }
  
@@ -1895,7 +1885,7 @@ index 2f8e401e743..7849c685b19 100644
      withTempView("v") {
        withSQLConf(
          SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "true",
-@@ -2239,7 +2278,7 @@ class AdaptiveQueryExecSuite
+@@ -2239,7 +2277,7 @@ class AdaptiveQueryExecSuite
                runAdaptiveAndVerifyResult(s"SELECT $repartition key1 FROM skewData1 " +
                  s"JOIN skewData2 ON key1 = key2 GROUP BY key1")
              val shuffles1 = collect(adaptive1) {
@@ -1904,7 +1894,7 @@ index 2f8e401e743..7849c685b19 100644
              }
              assert(shuffles1.size == 3)
              // shuffles1.head is the top-level shuffle under the Aggregate operator
-@@ -2252,7 +2291,7 @@ class AdaptiveQueryExecSuite
+@@ -2252,7 +2290,7 @@ class AdaptiveQueryExecSuite
                runAdaptiveAndVerifyResult(s"SELECT $repartition key1 FROM skewData1 " +
                  s"JOIN skewData2 ON key1 = key2")
              val shuffles2 = collect(adaptive2) {
@@ -1913,7 +1903,7 @@ index 2f8e401e743..7849c685b19 100644
              }
              if (hasRequiredDistribution) {
                assert(shuffles2.size == 3)
-@@ -2286,7 +2325,8 @@ class AdaptiveQueryExecSuite
+@@ -2286,7 +2324,8 @@ class AdaptiveQueryExecSuite
      }
    }
  
@@ -1923,7 +1913,7 @@ index 2f8e401e743..7849c685b19 100644
      CostEvaluator.instantiate(
        classOf[SimpleShuffleSortCostEvaluator].getCanonicalName, spark.sparkContext.getConf)
      intercept[IllegalArgumentException] {
-@@ -2452,6 +2492,7 @@ class AdaptiveQueryExecSuite
+@@ -2452,6 +2491,7 @@ class AdaptiveQueryExecSuite
            val (_, adaptive) = runAdaptiveAndVerifyResult(query)
            assert(adaptive.collect {
              case sort: SortExec => sort
@@ -1931,7 +1921,7 @@ index 2f8e401e743..7849c685b19 100644
            }.size == 1)
            val read = collect(adaptive) {
              case read: AQEShuffleReadExec => read
-@@ -2469,7 +2510,8 @@ class AdaptiveQueryExecSuite
+@@ -2469,7 +2509,8 @@ class AdaptiveQueryExecSuite
      }
    }
  
@@ -1941,7 +1931,7 @@ index 2f8e401e743..7849c685b19 100644
      withTempView("v") {
        withSQLConf(
          SQLConf.ADAPTIVE_OPTIMIZE_SKEWS_IN_REBALANCE_PARTITIONS_ENABLED.key -> "true",
-@@ -2581,7 +2623,7 @@ class AdaptiveQueryExecSuite
+@@ -2581,7 +2622,7 @@ class AdaptiveQueryExecSuite
            runAdaptiveAndVerifyResult("SELECT key1 FROM skewData1 JOIN skewData2 ON key1 = key2 " +
              "JOIN skewData3 ON value2 = value3")
          val shuffles1 = collect(adaptive1) {
@@ -1950,7 +1940,7 @@ index 2f8e401e743..7849c685b19 100644
          }
          assert(shuffles1.size == 4)
          val smj1 = findTopLevelSortMergeJoin(adaptive1)
-@@ -2592,7 +2634,7 @@ class AdaptiveQueryExecSuite
+@@ -2592,7 +2633,7 @@ class AdaptiveQueryExecSuite
            runAdaptiveAndVerifyResult("SELECT key1 FROM skewData1 JOIN skewData2 ON key1 = key2 " +
              "JOIN skewData3 ON value1 = value3")
          val shuffles2 = collect(adaptive2) {
@@ -1959,7 +1949,7 @@ index 2f8e401e743..7849c685b19 100644
          }
          assert(shuffles2.size == 4)
          val smj2 = findTopLevelSortMergeJoin(adaptive2)
-@@ -2850,6 +2892,7 @@ class AdaptiveQueryExecSuite
+@@ -2850,6 +2891,7 @@ class AdaptiveQueryExecSuite
          }.size == (if (firstAccess) 1 else 0))
          assert(collect(initialExecutedPlan) {
            case s: SortExec => s
@@ -1967,7 +1957,7 @@ index 2f8e401e743..7849c685b19 100644
          }.size == (if (firstAccess) 2 else 0))
          assert(collect(initialExecutedPlan) {
            case i: InMemoryTableScanLike => i
-@@ -2938,7 +2981,8 @@ class AdaptiveQueryExecSuite
+@@ -2938,7 +2980,8 @@ class AdaptiveQueryExecSuite
      }
    }
  
@@ -1977,7 +1967,7 @@ index 2f8e401e743..7849c685b19 100644
      val emptyDf = spark.range(1).where("false")
      val aggDf1 = emptyDf.agg(sum("id").as("id")).withColumn("name", lit("df1"))
      val aggDf2 = emptyDf.agg(sum("id").as("id")).withColumn("name", lit("df2"))
-@@ -2980,7 +3024,9 @@ class AdaptiveQueryExecSuite
+@@ -2980,7 +3023,9 @@ class AdaptiveQueryExecSuite
  
        val plan = df.queryExecution.executedPlan.asInstanceOf[AdaptiveSparkPlanExec]
        assert(plan.inputPlan.isInstanceOf[TakeOrderedAndProjectExec])
diff --git a/dev/diffs/4.0.4.diff b/dev/diffs/4.0.4.diff
index 38ec78543aa..7b06f566c22 100644
--- a/dev/diffs/4.0.4.diff
+++ b/dev/diffs/4.0.4.diff
@@ -2208,7 +2208,7 @@ index a3cfdc5a240..3793b6191bf 100644
        })
        checkAnswer(distinctWithId, Seq(Row(1, 0), Row(1, 0)))
 diff --git a/sql/core/src/test/scala/org/apache/spark/sql/execution/adaptive/AdaptiveQueryExecSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/execution/adaptive/AdaptiveQueryExecSuite.scala
-index fb8fab6a80f..403eb411920 100644
+index fb8fab6a80f..e5f2c3c5f9a 100644
 --- a/sql/core/src/test/scala/org/apache/spark/sql/execution/adaptive/AdaptiveQueryExecSuite.scala
 +++ b/sql/core/src/test/scala/org/apache/spark/sql/execution/adaptive/AdaptiveQueryExecSuite.scala
 @@ -28,12 +28,14 @@ import org.apache.spark.SparkException
@@ -2338,17 +2338,7 @@ index fb8fab6a80f..403eb411920 100644
      withSQLConf(
        SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "true",
        SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "80",
-@@ -301,7 +327,8 @@ class AdaptiveQueryExecSuite
-       val localReads = collect(adaptivePlan) {
-         case read: AQEShuffleReadExec if read.isLocalRead => read
-       }
--      assert(localReads.length == 2)
-+      // Comet shuffle changes shuffle metrics
-+      assert(localReads.length == 1)
-       val localShuffleRDD0 = localReads(0).execute().asInstanceOf[ShuffledRowRDD]
-       val localShuffleRDD1 = localReads(1).execute().asInstanceOf[ShuffledRowRDD]
-       // the final parallelism is math.max(1, numReduces / numMappers): math.max(1, 5/2) = 2
-@@ -326,7 +353,9 @@ class AdaptiveQueryExecSuite
+@@ -326,7 +352,9 @@ class AdaptiveQueryExecSuite
            .groupBy($"a").count()
          checkAnswer(testDf, Seq())
          val plan = testDf.queryExecution.executedPlan
@@ -2359,7 +2349,7 @@ index fb8fab6a80f..403eb411920 100644
          val coalescedReads = collect(plan) {
            case r: AQEShuffleReadExec => r
          }
-@@ -340,7 +369,9 @@ class AdaptiveQueryExecSuite
+@@ -340,7 +368,9 @@ class AdaptiveQueryExecSuite
            .groupBy($"a").count()
          checkAnswer(testDf, Seq())
          val plan = testDf.queryExecution.executedPlan
@@ -2370,7 +2360,7 @@ index fb8fab6a80f..403eb411920 100644
          val coalescedReads = collect(plan) {
            case r: AQEShuffleReadExec => r
          }
-@@ -350,7 +381,7 @@ class AdaptiveQueryExecSuite
+@@ -350,7 +380,7 @@ class AdaptiveQueryExecSuite
      }
    }
  
@@ -2379,7 +2369,7 @@ index fb8fab6a80f..403eb411920 100644
      withSQLConf(
          SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "true",
          SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "80") {
-@@ -365,7 +396,7 @@ class AdaptiveQueryExecSuite
+@@ -365,7 +395,7 @@ class AdaptiveQueryExecSuite
      }
    }
  
@@ -2388,7 +2378,7 @@ index fb8fab6a80f..403eb411920 100644
      withSQLConf(
          SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "true",
          SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "80") {
-@@ -381,7 +412,7 @@ class AdaptiveQueryExecSuite
+@@ -381,7 +411,7 @@ class AdaptiveQueryExecSuite
      }
    }
  
@@ -2397,7 +2387,7 @@ index fb8fab6a80f..403eb411920 100644
      withSQLConf(
          SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "true",
          SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "80") {
-@@ -426,7 +457,7 @@ class AdaptiveQueryExecSuite
+@@ -426,7 +456,7 @@ class AdaptiveQueryExecSuite
      }
    }
  
@@ -2406,7 +2396,7 @@ index fb8fab6a80f..403eb411920 100644
      withSQLConf(
          SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "true",
          SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "80") {
-@@ -471,7 +502,7 @@ class AdaptiveQueryExecSuite
+@@ -471,7 +501,7 @@ class AdaptiveQueryExecSuite
      }
    }
  
@@ -2415,7 +2405,7 @@ index fb8fab6a80f..403eb411920 100644
      withSQLConf(
          SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "true",
          SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "500") {
-@@ -517,7 +548,7 @@ class AdaptiveQueryExecSuite
+@@ -517,7 +547,7 @@ class AdaptiveQueryExecSuite
      }
    }
  
@@ -2424,7 +2414,7 @@ index fb8fab6a80f..403eb411920 100644
      withSQLConf(
          SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "true",
          SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "80") {
-@@ -536,7 +567,7 @@ class AdaptiveQueryExecSuite
+@@ -536,7 +566,7 @@ class AdaptiveQueryExecSuite
      }
    }
  
@@ -2433,7 +2423,7 @@ index fb8fab6a80f..403eb411920 100644
      withSQLConf(
          SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "true",
          SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "80") {
-@@ -567,7 +598,9 @@ class AdaptiveQueryExecSuite
+@@ -567,7 +597,9 @@ class AdaptiveQueryExecSuite
        assert(smj.size == 1)
        val bhj = findTopLevelBroadcastHashJoin(adaptivePlan)
        assert(bhj.size == 1)
@@ -2444,7 +2434,7 @@ index fb8fab6a80f..403eb411920 100644
        // Even with local shuffle read, the query stage reuse can also work.
        val ex = findReusedExchange(adaptivePlan)
        assert(ex.nonEmpty)
-@@ -588,7 +621,9 @@ class AdaptiveQueryExecSuite
+@@ -588,7 +620,9 @@ class AdaptiveQueryExecSuite
        assert(smj.size == 1)
        val bhj = findTopLevelBroadcastHashJoin(adaptivePlan)
        assert(bhj.size == 1)
@@ -2455,7 +2445,7 @@ index fb8fab6a80f..403eb411920 100644
        // Even with local shuffle read, the query stage reuse can also work.
        val ex = findReusedExchange(adaptivePlan)
        assert(ex.isEmpty)
-@@ -597,7 +632,8 @@ class AdaptiveQueryExecSuite
+@@ -597,7 +631,8 @@ class AdaptiveQueryExecSuite
      }
    }
  
@@ -2465,7 +2455,7 @@ index fb8fab6a80f..403eb411920 100644
      withSQLConf(
          SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "true",
          SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "20000000",
-@@ -692,7 +728,8 @@ class AdaptiveQueryExecSuite
+@@ -692,7 +727,8 @@ class AdaptiveQueryExecSuite
        val bhj = findTopLevelBroadcastHashJoin(adaptivePlan)
        assert(bhj.size == 1)
        // There is still a SMJ, and its two shuffles can't apply local read.
@@ -2475,7 +2465,7 @@ index fb8fab6a80f..403eb411920 100644
      }
    }
  
-@@ -814,7 +851,8 @@ class AdaptiveQueryExecSuite
+@@ -814,7 +850,8 @@ class AdaptiveQueryExecSuite
      }
    }
  
@@ -2485,7 +2475,7 @@ index fb8fab6a80f..403eb411920 100644
      Seq("SHUFFLE_MERGE", "SHUFFLE_HASH").foreach { joinHint =>
        def getJoinNode(plan: SparkPlan): Seq[ShuffledJoin] = if (joinHint == "SHUFFLE_MERGE") {
          findTopLevelSortMergeJoin(plan)
-@@ -1096,7 +1134,8 @@ class AdaptiveQueryExecSuite
+@@ -1096,7 +1133,8 @@ class AdaptiveQueryExecSuite
      }
    }
  
@@ -2495,7 +2485,7 @@ index fb8fab6a80f..403eb411920 100644
      withSQLConf(SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "true") {
        val (_, adaptivePlan) = runAdaptiveAndVerifyResult(
          "SELECT key FROM testData GROUP BY key")
-@@ -1662,13 +1701,13 @@ class AdaptiveQueryExecSuite
+@@ -1662,13 +1700,13 @@ class AdaptiveQueryExecSuite
        val (plan1, adaptivePlan1) = runAdaptiveAndVerifyResult(
          "SELECT key FROM testData WHERE key = 0 ORDER BY key, value")
        assert(findTopLevelSort(plan1).size == 1)
@@ -2511,7 +2501,7 @@ index fb8fab6a80f..403eb411920 100644
      }
    }
  
-@@ -1676,18 +1715,18 @@ class AdaptiveQueryExecSuite
+@@ -1676,18 +1714,18 @@ class AdaptiveQueryExecSuite
      withSQLConf(SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "true") {
        val (plan1, adaptivePlan1) = runAdaptiveAndVerifyResult(
          "SELECT key, count(*) FROM testData WHERE value = 'no_match' GROUP BY key")
@@ -2536,7 +2526,7 @@ index fb8fab6a80f..403eb411920 100644
      }
    }
  
-@@ -1708,7 +1747,7 @@ class AdaptiveQueryExecSuite
+@@ -1708,7 +1746,7 @@ class AdaptiveQueryExecSuite
            |""".stripMargin)
        checkNumUnion(plan1, 1)
        checkNumUnion(adaptivePlan1, 0)
@@ -2545,7 +2535,7 @@ index fb8fab6a80f..403eb411920 100644
  
        val (plan2, adaptivePlan2) = runAdaptiveAndVerifyResult(
          """
-@@ -1718,7 +1757,7 @@ class AdaptiveQueryExecSuite
+@@ -1718,7 +1756,7 @@ class AdaptiveQueryExecSuite
            |""".stripMargin)
        checkNumUnion(plan2, 1)
        checkNumUnion(adaptivePlan2, 0)
@@ -2554,7 +2544,7 @@ index fb8fab6a80f..403eb411920 100644
      }
    }
  
-@@ -1730,7 +1769,7 @@ class AdaptiveQueryExecSuite
+@@ -1730,7 +1768,7 @@ class AdaptiveQueryExecSuite
          val (_, adaptivePlan) = runAdaptiveAndVerifyResult(
            "SELECT id FROM v1 GROUP BY id DISTRIBUTE BY id")
          assert(collect(adaptivePlan) {
@@ -2563,7 +2553,7 @@ index fb8fab6a80f..403eb411920 100644
          }.length == 1)
        }
      }
-@@ -1810,7 +1849,8 @@ class AdaptiveQueryExecSuite
+@@ -1810,7 +1848,8 @@ class AdaptiveQueryExecSuite
      }
    }
  
@@ -2573,7 +2563,7 @@ index fb8fab6a80f..403eb411920 100644
      def hasRepartitionShuffle(plan: SparkPlan): Boolean = {
        find(plan) {
          case s: ShuffleExchangeLike =>
-@@ -1995,6 +2035,9 @@ class AdaptiveQueryExecSuite
+@@ -1995,6 +2034,9 @@ class AdaptiveQueryExecSuite
      def checkNoCoalescePartitions(ds: Dataset[Row], origin: ShuffleOrigin): Unit = {
        assert(collect(ds.queryExecution.executedPlan) {
          case s: ShuffleExchangeExec if s.shuffleOrigin == origin && s.numPartitions == 2 => s
@@ -2583,7 +2573,7 @@ index fb8fab6a80f..403eb411920 100644
        }.size == 1)
        ds.collect()
        val plan = ds.queryExecution.executedPlan
-@@ -2003,6 +2046,9 @@ class AdaptiveQueryExecSuite
+@@ -2003,6 +2045,9 @@ class AdaptiveQueryExecSuite
        }.isEmpty)
        assert(collect(plan) {
          case s: ShuffleExchangeExec if s.shuffleOrigin == origin && s.numPartitions == 2 => s
@@ -2593,7 +2583,7 @@ index fb8fab6a80f..403eb411920 100644
        }.size == 1)
        checkAnswer(ds, testData)
      }
-@@ -2159,7 +2205,8 @@ class AdaptiveQueryExecSuite
+@@ -2159,7 +2204,8 @@ class AdaptiveQueryExecSuite
      }
    }
  
@@ -2603,7 +2593,7 @@ index fb8fab6a80f..403eb411920 100644
      withTempView("t1", "t2") {
        def checkJoinStrategy(shouldShuffleHashJoin: Boolean): Unit = {
          Seq("100", "100000").foreach { size =>
-@@ -2245,7 +2292,8 @@ class AdaptiveQueryExecSuite
+@@ -2245,7 +2291,8 @@ class AdaptiveQueryExecSuite
      }
    }
  
@@ -2613,7 +2603,7 @@ index fb8fab6a80f..403eb411920 100644
      withTempView("v") {
        withSQLConf(
          SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "true",
-@@ -2344,7 +2392,7 @@ class AdaptiveQueryExecSuite
+@@ -2344,7 +2391,7 @@ class AdaptiveQueryExecSuite
                runAdaptiveAndVerifyResult(s"SELECT $repartition key1 FROM skewData1 " +
                  s"JOIN skewData2 ON key1 = key2 GROUP BY key1")
              val shuffles1 = collect(adaptive1) {
@@ -2622,7 +2612,7 @@ index fb8fab6a80f..403eb411920 100644
              }
              assert(shuffles1.size == 3)
              // shuffles1.head is the top-level shuffle under the Aggregate operator
-@@ -2357,7 +2405,7 @@ class AdaptiveQueryExecSuite
+@@ -2357,7 +2404,7 @@ class AdaptiveQueryExecSuite
                runAdaptiveAndVerifyResult(s"SELECT $repartition key1 FROM skewData1 " +
                  s"JOIN skewData2 ON key1 = key2")
              val shuffles2 = collect(adaptive2) {
@@ -2631,7 +2621,7 @@ index fb8fab6a80f..403eb411920 100644
              }
              if (hasRequiredDistribution) {
                assert(shuffles2.size == 3)
-@@ -2391,7 +2439,8 @@ class AdaptiveQueryExecSuite
+@@ -2391,7 +2438,8 @@ class AdaptiveQueryExecSuite
      }
    }
  
@@ -2641,7 +2631,7 @@ index fb8fab6a80f..403eb411920 100644
      CostEvaluator.instantiate(
        classOf[SimpleShuffleSortCostEvaluator].getCanonicalName, spark.sparkContext.getConf)
      intercept[IllegalArgumentException] {
-@@ -2557,6 +2606,7 @@ class AdaptiveQueryExecSuite
+@@ -2557,6 +2605,7 @@ class AdaptiveQueryExecSuite
            val (_, adaptive) = runAdaptiveAndVerifyResult(query)
            assert(adaptive.collect {
              case sort: SortExec => sort
@@ -2649,7 +2639,7 @@ index fb8fab6a80f..403eb411920 100644
            }.size == 1)
            val read = collect(adaptive) {
              case read: AQEShuffleReadExec => read
-@@ -2574,7 +2624,8 @@ class AdaptiveQueryExecSuite
+@@ -2574,7 +2623,8 @@ class AdaptiveQueryExecSuite
      }
    }
  
@@ -2659,7 +2649,7 @@ index fb8fab6a80f..403eb411920 100644
      withTempView("v") {
        withSQLConf(
          SQLConf.ADAPTIVE_OPTIMIZE_SKEWS_IN_REBALANCE_PARTITIONS_ENABLED.key -> "true",
-@@ -2686,7 +2737,7 @@ class AdaptiveQueryExecSuite
+@@ -2686,7 +2736,7 @@ class AdaptiveQueryExecSuite
            runAdaptiveAndVerifyResult("SELECT key1 FROM skewData1 JOIN skewData2 ON key1 = key2 " +
              "JOIN skewData3 ON value2 = value3")
          val shuffles1 = collect(adaptive1) {
@@ -2668,7 +2658,7 @@ index fb8fab6a80f..403eb411920 100644
          }
          assert(shuffles1.size == 4)
          val smj1 = findTopLevelSortMergeJoin(adaptive1)
-@@ -2697,7 +2748,7 @@ class AdaptiveQueryExecSuite
+@@ -2697,7 +2747,7 @@ class AdaptiveQueryExecSuite
            runAdaptiveAndVerifyResult("SELECT key1 FROM skewData1 JOIN skewData2 ON key1 = key2 " +
              "JOIN skewData3 ON value1 = value3")
          val shuffles2 = collect(adaptive2) {
@@ -2677,7 +2667,7 @@ index fb8fab6a80f..403eb411920 100644
          }
          assert(shuffles2.size == 4)
          val smj2 = findTopLevelSortMergeJoin(adaptive2)
-@@ -2955,6 +3006,7 @@ class AdaptiveQueryExecSuite
+@@ -2955,6 +3005,7 @@ class AdaptiveQueryExecSuite
          }.size == (if (firstAccess) 1 else 0))
          assert(collect(initialExecutedPlan) {
            case s: SortExec => s
@@ -2685,7 +2675,7 @@ index fb8fab6a80f..403eb411920 100644
          }.size == (if (firstAccess) 2 else 0))
          assert(collect(initialExecutedPlan) {
            case i: InMemoryTableScanLike => i
-@@ -2967,6 +3019,7 @@ class AdaptiveQueryExecSuite
+@@ -2967,6 +3018,7 @@ class AdaptiveQueryExecSuite
          }.isEmpty)
          assert(collect(finalExecutedPlan) {
            case s: SortExec => s
@@ -2693,7 +2683,7 @@ index fb8fab6a80f..403eb411920 100644
          }.isEmpty)
          assert(collect(initialExecutedPlan) {
            case i: InMemoryTableScanLike => i
-@@ -3048,7 +3101,8 @@ class AdaptiveQueryExecSuite
+@@ -3048,7 +3100,8 @@ class AdaptiveQueryExecSuite
      }
    }
  
@@ -2703,7 +2693,7 @@ index fb8fab6a80f..403eb411920 100644
      val emptyDf = spark.range(1).where("false")
      val aggDf1 = emptyDf.agg(sum("id").as("id")).withColumn("name", lit("df1"))
      val aggDf2 = emptyDf.agg(sum("id").as("id")).withColumn("name", lit("df2"))
-@@ -3138,7 +3192,8 @@ class AdaptiveQueryExecSuite
+@@ -3138,7 +3191,8 @@ class AdaptiveQueryExecSuite
  
        val plan = df.queryExecution.executedPlan.asInstanceOf[AdaptiveSparkPlanExec]
        assert(plan.inputPlan.isInstanceOf[TakeOrderedAndProjectExec])
diff --git a/dev/diffs/4.1.3.diff b/dev/diffs/4.1.3.diff
index a8d9383ced1..6baa84d1531 100644
--- a/dev/diffs/4.1.3.diff
+++ b/dev/diffs/4.1.3.diff
@@ -2328,7 +2328,7 @@ index a3cfdc5a240..3793b6191bf 100644
        })
        checkAnswer(distinctWithId, Seq(Row(1, 0), Row(1, 0)))
 diff --git a/sql/core/src/test/scala/org/apache/spark/sql/execution/adaptive/AdaptiveQueryExecSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/execution/adaptive/AdaptiveQueryExecSuite.scala
-index 188a28ff1c0..8fdccf31749 100644
+index 188a28ff1c0..c7e622d9982 100644
 --- a/sql/core/src/test/scala/org/apache/spark/sql/execution/adaptive/AdaptiveQueryExecSuite.scala
 +++ b/sql/core/src/test/scala/org/apache/spark/sql/execution/adaptive/AdaptiveQueryExecSuite.scala
 @@ -27,12 +27,14 @@ import org.apache.spark.SparkException
@@ -2458,17 +2458,7 @@ index 188a28ff1c0..8fdccf31749 100644
      withSQLConf(
        SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "true",
        SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "80",
-@@ -303,7 +329,8 @@ class AdaptiveQueryExecSuite
-       val localReads = collect(adaptivePlan) {
-         case read: AQEShuffleReadExec if read.isLocalRead => read
-       }
--      assert(localReads.length == 2)
-+      // Comet shuffle changes shuffle metrics
-+      assert(localReads.length == 1)
-       val localShuffleRDD0 = localReads(0).execute().asInstanceOf[ShuffledRowRDD]
-       val localShuffleRDD1 = localReads(1).execute().asInstanceOf[ShuffledRowRDD]
-       // the final parallelism is math.max(1, numReduces / numMappers): math.max(1, 5/2) = 2
-@@ -328,7 +355,9 @@ class AdaptiveQueryExecSuite
+@@ -328,7 +354,9 @@ class AdaptiveQueryExecSuite
            .groupBy($"a").count()
          checkAnswer(testDf, Seq())
          val plan = testDf.queryExecution.executedPlan
@@ -2479,7 +2469,7 @@ index 188a28ff1c0..8fdccf31749 100644
          val coalescedReads = collect(plan) {
            case r: AQEShuffleReadExec => r
          }
-@@ -342,7 +371,9 @@ class AdaptiveQueryExecSuite
+@@ -342,7 +370,9 @@ class AdaptiveQueryExecSuite
            .groupBy($"a").count()
          checkAnswer(testDf, Seq())
          val plan = testDf.queryExecution.executedPlan
@@ -2490,7 +2480,7 @@ index 188a28ff1c0..8fdccf31749 100644
          val coalescedReads = collect(plan) {
            case r: AQEShuffleReadExec => r
          }
-@@ -352,7 +383,7 @@ class AdaptiveQueryExecSuite
+@@ -352,7 +382,7 @@ class AdaptiveQueryExecSuite
      }
    }
  
@@ -2499,7 +2489,7 @@ index 188a28ff1c0..8fdccf31749 100644
      withSQLConf(
          SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "true",
          SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "80") {
-@@ -367,7 +398,7 @@ class AdaptiveQueryExecSuite
+@@ -367,7 +397,7 @@ class AdaptiveQueryExecSuite
      }
    }
  
@@ -2508,7 +2498,7 @@ index 188a28ff1c0..8fdccf31749 100644
      withSQLConf(
          SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "true",
          SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "80") {
-@@ -383,7 +414,7 @@ class AdaptiveQueryExecSuite
+@@ -383,7 +413,7 @@ class AdaptiveQueryExecSuite
      }
    }
  
@@ -2517,7 +2507,7 @@ index 188a28ff1c0..8fdccf31749 100644
      withSQLConf(
          SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "true",
          SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "80") {
-@@ -428,7 +459,7 @@ class AdaptiveQueryExecSuite
+@@ -428,7 +458,7 @@ class AdaptiveQueryExecSuite
      }
    }
  
@@ -2526,7 +2516,7 @@ index 188a28ff1c0..8fdccf31749 100644
      withSQLConf(
          SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "true",
          SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "80") {
-@@ -473,7 +504,7 @@ class AdaptiveQueryExecSuite
+@@ -473,7 +503,7 @@ class AdaptiveQueryExecSuite
      }
    }
  
@@ -2535,7 +2525,7 @@ index 188a28ff1c0..8fdccf31749 100644
      withSQLConf(
          SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "true",
          SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "500") {
-@@ -519,7 +550,7 @@ class AdaptiveQueryExecSuite
+@@ -519,7 +549,7 @@ class AdaptiveQueryExecSuite
      }
    }
  
@@ -2544,7 +2534,7 @@ index 188a28ff1c0..8fdccf31749 100644
      withSQLConf(
          SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "true",
          SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "80") {
-@@ -538,7 +569,7 @@ class AdaptiveQueryExecSuite
+@@ -538,7 +568,7 @@ class AdaptiveQueryExecSuite
      }
    }
  
@@ -2553,7 +2543,7 @@ index 188a28ff1c0..8fdccf31749 100644
      withSQLConf(
          SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "true",
          SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "80") {
-@@ -569,7 +600,9 @@ class AdaptiveQueryExecSuite
+@@ -569,7 +599,9 @@ class AdaptiveQueryExecSuite
        assert(smj.size == 1)
        val bhj = findTopLevelBroadcastHashJoin(adaptivePlan)
        assert(bhj.size == 1)
@@ -2564,7 +2554,7 @@ index 188a28ff1c0..8fdccf31749 100644
        // Even with local shuffle read, the query stage reuse can also work.
        val ex = findReusedExchange(adaptivePlan)
        assert(ex.nonEmpty)
-@@ -590,7 +623,9 @@ class AdaptiveQueryExecSuite
+@@ -590,7 +622,9 @@ class AdaptiveQueryExecSuite
        assert(smj.size == 1)
        val bhj = findTopLevelBroadcastHashJoin(adaptivePlan)
        assert(bhj.size == 1)
@@ -2575,7 +2565,7 @@ index 188a28ff1c0..8fdccf31749 100644
        // Even with local shuffle read, the query stage reuse can also work.
        val ex = findReusedExchange(adaptivePlan)
        assert(ex.isEmpty)
-@@ -599,7 +634,8 @@ class AdaptiveQueryExecSuite
+@@ -599,7 +633,8 @@ class AdaptiveQueryExecSuite
      }
    }
  
@@ -2585,7 +2575,7 @@ index 188a28ff1c0..8fdccf31749 100644
      withSQLConf(
          SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "true",
          SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "20000000",
-@@ -694,7 +730,8 @@ class AdaptiveQueryExecSuite
+@@ -694,7 +729,8 @@ class AdaptiveQueryExecSuite
        val bhj = findTopLevelBroadcastHashJoin(adaptivePlan)
        assert(bhj.size == 1)
        // There is still a SMJ, and its two shuffles can't apply local read.
@@ -2595,7 +2585,7 @@ index 188a28ff1c0..8fdccf31749 100644
      }
    }
  
-@@ -822,7 +859,8 @@ class AdaptiveQueryExecSuite
+@@ -822,7 +858,8 @@ class AdaptiveQueryExecSuite
      }
    }
  
@@ -2605,7 +2595,7 @@ index 188a28ff1c0..8fdccf31749 100644
      Seq("SHUFFLE_MERGE", "SHUFFLE_HASH").foreach { joinHint =>
        def getJoinNode(plan: SparkPlan): Seq[ShuffledJoin] = if (joinHint == "SHUFFLE_MERGE") {
          findTopLevelSortMergeJoin(plan)
-@@ -1266,7 +1304,8 @@ class AdaptiveQueryExecSuite
+@@ -1266,7 +1303,8 @@ class AdaptiveQueryExecSuite
      }
    }
  
@@ -2615,7 +2605,7 @@ index 188a28ff1c0..8fdccf31749 100644
      withSQLConf(SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "true") {
        val (_, adaptivePlan) = runAdaptiveAndVerifyResult(
          "SELECT key FROM testData GROUP BY key")
-@@ -1832,13 +1871,13 @@ class AdaptiveQueryExecSuite
+@@ -1832,13 +1870,13 @@ class AdaptiveQueryExecSuite
        val (plan1, adaptivePlan1) = runAdaptiveAndVerifyResult(
          "SELECT key FROM testData WHERE key = 0 ORDER BY key, value")
        assert(findTopLevelSort(plan1).size == 1)
@@ -2631,7 +2621,7 @@ index 188a28ff1c0..8fdccf31749 100644
      }
    }
  
-@@ -1846,18 +1885,18 @@ class AdaptiveQueryExecSuite
+@@ -1846,18 +1884,18 @@ class AdaptiveQueryExecSuite
      withSQLConf(SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "true") {
        val (plan1, adaptivePlan1) = runAdaptiveAndVerifyResult(
          "SELECT key, count(*) FROM testData WHERE value = 'no_match' GROUP BY key")
@@ -2656,7 +2646,7 @@ index 188a28ff1c0..8fdccf31749 100644
      }
    }
  
-@@ -1878,7 +1917,7 @@ class AdaptiveQueryExecSuite
+@@ -1878,7 +1916,7 @@ class AdaptiveQueryExecSuite
            |""".stripMargin)
        checkNumUnion(plan1, 1)
        checkNumUnion(adaptivePlan1, 0)
@@ -2665,7 +2655,7 @@ index 188a28ff1c0..8fdccf31749 100644
  
        val (plan2, adaptivePlan2) = runAdaptiveAndVerifyResult(
          """
-@@ -1888,7 +1927,7 @@ class AdaptiveQueryExecSuite
+@@ -1888,7 +1926,7 @@ class AdaptiveQueryExecSuite
            |""".stripMargin)
        checkNumUnion(plan2, 1)
        checkNumUnion(adaptivePlan2, 0)
@@ -2674,7 +2664,7 @@ index 188a28ff1c0..8fdccf31749 100644
      }
    }
  
-@@ -1900,7 +1939,7 @@ class AdaptiveQueryExecSuite
+@@ -1900,7 +1938,7 @@ class AdaptiveQueryExecSuite
          val (_, adaptivePlan) = runAdaptiveAndVerifyResult(
            "SELECT id FROM v1 GROUP BY id DISTRIBUTE BY id")
          assert(collect(adaptivePlan) {
@@ -2683,7 +2673,7 @@ index 188a28ff1c0..8fdccf31749 100644
          }.length == 1)
        }
      }
-@@ -1981,7 +2020,8 @@ class AdaptiveQueryExecSuite
+@@ -1981,7 +2019,8 @@ class AdaptiveQueryExecSuite
      }
    }
  
@@ -2693,7 +2683,7 @@ index 188a28ff1c0..8fdccf31749 100644
      def hasRepartitionShuffle(plan: SparkPlan): Boolean = {
        find(plan) {
          case s: ShuffleExchangeLike =>
-@@ -2168,6 +2208,9 @@ class AdaptiveQueryExecSuite
+@@ -2168,6 +2207,9 @@ class AdaptiveQueryExecSuite
      def checkNoCoalescePartitions(ds: Dataset[Row], origin: ShuffleOrigin): Unit = {
        assert(collect(ds.queryExecution.executedPlan) {
          case s: ShuffleExchangeExec if s.shuffleOrigin == origin && s.numPartitions == 2 => s
@@ -2703,7 +2693,7 @@ index 188a28ff1c0..8fdccf31749 100644
        }.size == 1)
        ds.collect()
        val plan = ds.queryExecution.executedPlan
-@@ -2176,6 +2219,9 @@ class AdaptiveQueryExecSuite
+@@ -2176,6 +2218,9 @@ class AdaptiveQueryExecSuite
        }.isEmpty)
        assert(collect(plan) {
          case s: ShuffleExchangeExec if s.shuffleOrigin == origin && s.numPartitions == 2 => s
@@ -2713,7 +2703,7 @@ index 188a28ff1c0..8fdccf31749 100644
        }.size == 1)
        checkAnswer(ds, testData)
      }
-@@ -2340,7 +2386,8 @@ class AdaptiveQueryExecSuite
+@@ -2340,7 +2385,8 @@ class AdaptiveQueryExecSuite
      }
    }
  
@@ -2723,7 +2713,7 @@ index 188a28ff1c0..8fdccf31749 100644
      withTempView("t1", "t2") {
        def checkJoinStrategy(shouldShuffleHashJoin: Boolean): Unit = {
          Seq("100", "100000").foreach { size =>
-@@ -2426,7 +2473,8 @@ class AdaptiveQueryExecSuite
+@@ -2426,7 +2472,8 @@ class AdaptiveQueryExecSuite
      }
    }
  
@@ -2733,7 +2723,7 @@ index 188a28ff1c0..8fdccf31749 100644
      withTempView("v") {
        withSQLConf(
          SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "true",
-@@ -2525,7 +2573,7 @@ class AdaptiveQueryExecSuite
+@@ -2525,7 +2572,7 @@ class AdaptiveQueryExecSuite
                runAdaptiveAndVerifyResult(s"SELECT $repartition key1 FROM skewData1 " +
                  s"JOIN skewData2 ON key1 = key2 GROUP BY key1")
              val shuffles1 = collect(adaptive1) {
@@ -2742,7 +2732,7 @@ index 188a28ff1c0..8fdccf31749 100644
              }
              assert(shuffles1.size == 3)
              // shuffles1.head is the top-level shuffle under the Aggregate operator
-@@ -2538,7 +2586,7 @@ class AdaptiveQueryExecSuite
+@@ -2538,7 +2585,7 @@ class AdaptiveQueryExecSuite
                runAdaptiveAndVerifyResult(s"SELECT $repartition key1 FROM skewData1 " +
                  s"JOIN skewData2 ON key1 = key2")
              val shuffles2 = collect(adaptive2) {
@@ -2751,7 +2741,7 @@ index 188a28ff1c0..8fdccf31749 100644
              }
              if (hasRequiredDistribution) {
                assert(shuffles2.size == 3)
-@@ -2572,7 +2620,8 @@ class AdaptiveQueryExecSuite
+@@ -2572,7 +2619,8 @@ class AdaptiveQueryExecSuite
      }
    }
  
@@ -2761,7 +2751,7 @@ index 188a28ff1c0..8fdccf31749 100644
      CostEvaluator.instantiate(
        classOf[SimpleShuffleSortCostEvaluator].getCanonicalName, spark.sparkContext.getConf)
      intercept[IllegalArgumentException] {
-@@ -2738,6 +2787,7 @@ class AdaptiveQueryExecSuite
+@@ -2738,6 +2786,7 @@ class AdaptiveQueryExecSuite
            val (_, adaptive) = runAdaptiveAndVerifyResult(query)
            assert(adaptive.collect {
              case sort: SortExec => sort
@@ -2769,7 +2759,7 @@ index 188a28ff1c0..8fdccf31749 100644
            }.size == 1)
            val read = collect(adaptive) {
              case read: AQEShuffleReadExec => read
-@@ -2755,7 +2805,8 @@ class AdaptiveQueryExecSuite
+@@ -2755,7 +2804,8 @@ class AdaptiveQueryExecSuite
      }
    }
  
@@ -2779,7 +2769,7 @@ index 188a28ff1c0..8fdccf31749 100644
      withTempView("v") {
        withSQLConf(
          SQLConf.ADAPTIVE_OPTIMIZE_SKEWS_IN_REBALANCE_PARTITIONS_ENABLED.key -> "true",
-@@ -2867,7 +2918,7 @@ class AdaptiveQueryExecSuite
+@@ -2867,7 +2917,7 @@ class AdaptiveQueryExecSuite
            runAdaptiveAndVerifyResult("SELECT key1 FROM skewData1 JOIN skewData2 ON key1 = key2 " +
              "JOIN skewData3 ON value2 = value3")
          val shuffles1 = collect(adaptive1) {
@@ -2788,7 +2778,7 @@ index 188a28ff1c0..8fdccf31749 100644
          }
          assert(shuffles1.size == 4)
          val smj1 = findTopLevelSortMergeJoin(adaptive1)
-@@ -2878,7 +2929,7 @@ class AdaptiveQueryExecSuite
+@@ -2878,7 +2928,7 @@ class AdaptiveQueryExecSuite
            runAdaptiveAndVerifyResult("SELECT key1 FROM skewData1 JOIN skewData2 ON key1 = key2 " +
              "JOIN skewData3 ON value1 = value3")
          val shuffles2 = collect(adaptive2) {
@@ -2797,7 +2787,7 @@ index 188a28ff1c0..8fdccf31749 100644
          }
          assert(shuffles2.size == 4)
          val smj2 = findTopLevelSortMergeJoin(adaptive2)
-@@ -3136,6 +3187,7 @@ class AdaptiveQueryExecSuite
+@@ -3136,6 +3186,7 @@ class AdaptiveQueryExecSuite
          }.size == (if (firstAccess) 1 else 0))
          assert(collect(initialExecutedPlan) {
            case s: SortExec => s
@@ -2805,7 +2795,7 @@ index 188a28ff1c0..8fdccf31749 100644
          }.size == (if (firstAccess) 2 else 0))
          assert(collect(initialExecutedPlan) {
            case i: InMemoryTableScanLike => i
-@@ -3148,6 +3200,7 @@ class AdaptiveQueryExecSuite
+@@ -3148,6 +3199,7 @@ class AdaptiveQueryExecSuite
          }.isEmpty)
          assert(collect(finalExecutedPlan) {
            case s: SortExec => s
@@ -2813,7 +2803,7 @@ index 188a28ff1c0..8fdccf31749 100644
          }.isEmpty)
          assert(collect(initialExecutedPlan) {
            case i: InMemoryTableScanLike => i
-@@ -3229,7 +3282,8 @@ class AdaptiveQueryExecSuite
+@@ -3229,7 +3281,8 @@ class AdaptiveQueryExecSuite
      }
    }
  
@@ -2823,7 +2813,7 @@ index 188a28ff1c0..8fdccf31749 100644
      val emptyDf = spark.range(1).where("false")
      val aggDf1 = emptyDf.agg(sum("id").as("id")).withColumn("name", lit("df1"))
      val aggDf2 = emptyDf.agg(sum("id").as("id")).withColumn("name", lit("df2"))
-@@ -3319,7 +3373,8 @@ class AdaptiveQueryExecSuite
+@@ -3319,7 +3372,8 @@ class AdaptiveQueryExecSuite
  
        val plan = df.queryExecution.executedPlan.asInstanceOf[AdaptiveSparkPlanExec]
        assert(plan.inputPlan.isInstanceOf[TakeOrderedAndProjectExec])