[CARMEL-6055] Backport code-gen code for SortMergeJoin (#987)

* [CARMEL-6055] Backport code-gen code for SortMergeJoin

PR list:
 
* Cheng Su 2021/12/30, 12:12 PM [SPARK-37726][SQL] Add spill size metrics for sort merge join
* Cheng Su 2021/11/19, 12:36 PM [SPARK-37370][SQL] Add SQL configs to control newly added join code-gen in 3.3
* Cheng Su 2021/11/17, 9:48 PM [SPARK-37316][SQL] Add code-gen for existence sort merge join
* Cheng Su 2021/11/17, 10:44 AM [SPARK-37341][SQL] Avoid unnecessary buffer and copy in full outer sort merge join
* Cheng Su 2021/11/15, 7:34 PM [SPARK-35352][SQL] Add code-gen for full outer sort merge join
* Cheng Su 2021/11/3, 11:18 AM [SPARK-32567][SQL] Add code-gen for full outer shuffled hash join
* Cheng Su 2021/6/2, 2:01 PM [SPARK-35604][SQL] Fix condition check for FULL OUTER sort merge join
* Cheng Su 2021/5/27, 12:59 PM [SPARK-35351][SQL][FOLLOWUP] Avoid using `loaded` variable for LEFT ANTI SMJ code-gen
* Cheng Su 2021/5/18, 3:56 PM [SPARK-35351][SQL] Add code-gen for left anti sort merge join
* Cheng Su 2021/5/17, 1:49 AM [SPARK-35363][SQL][FOLLOWUP] Use fresh name for findNextJoinRows instead of hardcoding it
* Cheng Su 2021/5/13, 8:52 PM [SPARK-35350][SQL] Add code-gen for left semi sort merge join
* Cheng Su 2021/5/12, 10:10 PM [SPARK-35349][SQL] Add code-gen for left/right outer sort merge join
* Cheng Su 2021/5/11, 10:21 AM [SPARK-35363][SQL] Refactor sort merge join code-gen be agnostic to join type

* [CARMEL-6055] Backport code-gen code for SortMergeJoin

* Backport pr for BroadcastNestedLoopJoin

* Fix UT

* Fix UT

* Backport [CARMEL-3719] OOM in sort merge join

* Backport [CARMEL-3719] OOM in sort merge join

* fix ut

* Enable code-gen for full outer sort merge join by default

Author: wakun

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

@@ -1750,6 +1750,54 @@ object SQLConf {
       .booleanConf
       .createWithDefault(true)
 
+  val ENABLE_SORT_AGGREGATE_CODEGEN =
+    buildConf("spark.sql.codegen.aggregate.sortAggregate.enabled")
+      .internal()
+      .doc("When true, enable code-gen for sort aggregate.")
+      .version("3.3.0")
+      .booleanConf
+      .createWithDefault(true)
+
+  val ENABLE_FULL_OUTER_SHUFFLED_HASH_JOIN_CODEGEN =
+    buildConf("spark.sql.codegen.join.fullOuterShuffledHashJoin.enabled")
+      .internal()
+      .doc("When true, enable code-gen for FULL OUTER shuffled hash join.")
+      .version("3.3.0")
+      .booleanConf
+      .createWithDefault(true)
+
+  val ENABLE_OUTER_SORT_MERGE_JOIN_CODEGEN =
+    buildConf("spark.sql.codegen.join.outerSortMergeJoin.enabled")
+      .internal()
+      .doc("When true, enable code-gen for LEFT OUTER, RIGHT OUTER sort merge join")
+      .version("3.3.0")
+      .booleanConf
+      .createWithDefault(true)
+
+  val ENABLE_FULL_OUTER_SORT_MERGE_JOIN_CODEGEN =
+    buildConf("spark.sql.codegen.join.fullOuterSortMergeJoin.enabled")
+      .internal()
+      .doc("When true, enable code-gen for FULL OUTER sort merge join.")
+      .version("3.3.0")
+      .booleanConf
+      .createWithDefault(true)
+
+  val ENABLE_EXISTENCE_SORT_MERGE_JOIN_CODEGEN =
+    buildConf("spark.sql.codegen.join.existenceSortMergeJoin.enabled")
+      .internal()
+      .doc("When true, enable code-gen for Existence sort merge join.")
+      .version("3.3.0")
+      .booleanConf
+      .createWithDefault(true)
+
+  val ENABLE_SEMI_OR_ANTI_MERGE_JOIN_CODEGEN =
+    buildConf("spark.sql.codegen.join.semiOrAntiSortMergeJoin.enabled")
+      .internal()
+      .doc("When true, enable code-gen for semi or anti sort merge join")
+      .version("3.3.0")
+      .booleanConf
+      .createWithDefault(true)
+
   val MAX_NESTED_VIEW_DEPTH =
     buildConf("spark.sql.view.maxNestedViewDepth")
       .internal()

sql/core/src/main/scala/org/apache/spark/sql/execution/ExternalAppendOnlyUnsafeRowArray.scala

@@ -74,6 +74,7 @@ private[sql] class ExternalAppendOnlyUnsafeRowArray(
   }
 
   private var spillableArray: UnsafeExternalSorter = _
+  private var totalSpillBytes: Long = 0
   private var numRows = 0
 
   // A counter to keep track of total modifications done to this array since its creation.
@@ -86,6 +87,17 @@ private[sql] class ExternalAppendOnlyUnsafeRowArray(
 
   def isEmpty: Boolean = numRows == 0
 
+  /**
+   * Total number of bytes that has been spilled into disk so far.
+   */
+  def spillSize: Long = {
+    if (spillableArray != null) {
+      totalSpillBytes + spillableArray.getSpillSize
+    } else {
+      totalSpillBytes
+    }
+  }
+
   /**
    * Clears up resources (eg. memory) held by the backing storage
    */

sql/core/src/main/scala/org/apache/spark/sql/execution/WholeStageCodegenExec.scala

@@ -34,7 +34,7 @@ import org.apache.spark.sql.catalyst.plans.physical.Partitioning
 import org.apache.spark.sql.catalyst.rules.Rule
 import org.apache.spark.sql.execution.aggregate.HashAggregateExec
 import org.apache.spark.sql.execution.columnar.InMemoryTableScanExec
-import org.apache.spark.sql.execution.joins.{BroadcastHashJoinExec, ShuffledHashJoinExec, SortMergeJoinExec}
+import org.apache.spark.sql.execution.joins.{BroadcastHashJoinExec, BroadcastNestedLoopJoinExec, ShuffledHashJoinExec, SortMergeJoinExec}
 import org.apache.spark.sql.execution.metric.SQLMetrics
 import org.apache.spark.sql.internal.SQLConf
 import org.apache.spark.sql.types._
@@ -52,6 +52,7 @@ trait CodegenSupport extends SparkPlan {
     case _: BroadcastHashJoinExec => "bhj"
     case _: ShuffledHashJoinExec => "shj"
     case _: SortMergeJoinExec => "smj"
+    case _: BroadcastNestedLoopJoinExec => "bnlj"
     case _: RDDScanExec => "rdd"
     case _: DataSourceScanExec => "scan"
     case _: InMemoryTableScanExec => "memoryScan"