[CARMEL-6587] Support Generic Skew Join Patterns (#1281)

* [CARMEL-6587] Support Generic Skew Join Patterns

* fix ut

* fix ut

* fix ut

Author: xingchaozh

sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala

@@ -697,6 +697,15 @@ object SQLConf {
       .booleanConf
       .createWithDefault(true)
 
+  val GENERIC_SKEW_JOIN_ENABLED =
+    buildConf("spark.sql.adaptive.genericSkewJoin.enabled")
+      .doc(s"When true and '${SKEW_JOIN_ENABLED.key}' is true, Spark dynamically " +
+        "handles skew in sort-merge join by splitting (and replicating if needed) skewed " +
+        "partitions.")
+      .version("3.0.0")
+      .booleanConf
+      .createWithDefault(false)
+
   val SKEW_BROADCAST_HASH_JOIN_ENABLED =
     buildConf("spark.sql.adaptive.broadcastHashJoinSkew.enabled")
       .doc(s"When true and '${ADAPTIVE_EXECUTION_ENABLED.key}' is true, Spark dynamically " +
@@ -712,7 +721,7 @@ object SQLConf {
         "handles skew in broadcast hash join leveraging local shuffle read.")
       .version("3.0.0")
       .booleanConf
-      .createWithDefault(false)
+      .createWithDefault(true)
 
   val ALLOW_ADDITIONAL_SHUFFLE =
     buildConf("spark.sql.adaptive.allowAdditionalShuffle")

sql/core/src/main/scala/org/apache/spark/sql/execution/adaptive/AdaptiveSparkPlanExec.scala

@@ -99,6 +99,7 @@ case class AdaptiveSparkPlanExec(
     DisableUnnecessaryBucketedScan,
     AdjustScanPartitionSizeDynamically,
     OptimizeSkewedJoin, // ensureRequirements
+    OptimizeGenericSkewedJoin,
     EliminateSkewOptimzeIntroducedShuffle,
     removeRedundantSorts
   ) ++ context.session.sessionState.queryStagePrepRules

sql/core/src/main/scala/org/apache/spark/sql/execution/adaptive/CustomShuffleReaderExec.scala

@@ -146,12 +146,21 @@ case class CustomShuffleReaderExec private(
     driverAccumUpdates += (numPartitionsMetric.id -> partitionSpecs.length.toLong)
 
     if (hasSkewedPartition) {
-      val skewedMetric = metrics("numSkewedPartitions")
-      val numSkewedPartitions = partitionSpecs.collect {
+      val skewedSpecs = partitionSpecs.collect {
         case p: PartialReducerPartitionSpec => p.reducerIndex
-      }.distinct.length
-      skewedMetric.set(numSkewedPartitions)
-      driverAccumUpdates += (skewedMetric.id -> numSkewedPartitions.toLong)
+      }
+
+      val skewedPartitions = metrics("numSkewedPartitions")
+      val skewedSplits = metrics("numSkewedSplits")
+
+      val numSkewedPartitions = skewedSpecs.distinct.length
+      val numSplits = skewedSpecs.length
+
+      skewedPartitions.set(numSkewedPartitions)
+      driverAccumUpdates += (skewedPartitions.id -> numSkewedPartitions)
+
+      skewedSplits.set(numSplits)
+      driverAccumUpdates += (skewedSplits.id -> numSplits)
     }
 
     partitionDataSizes.foreach { dataSizes =>
@@ -178,7 +187,9 @@ case class CustomShuffleReaderExec private(
       } ++ {
         if (hasSkewedPartition) {
           Map("numSkewedPartitions" ->
-            SQLMetrics.createMetric(sparkContext, "number of skewed partitions"))
+            SQLMetrics.createMetric(sparkContext, "number of skewed partitions"),
+            "numSkewedSplits" ->
+              SQLMetrics.createMetric(sparkContext, "number of skewed partition splits"))
         } else {
           Map.empty
         }