fix: Do not replace SMJ with HJ for LeftSemi

Author: comphead

spark/src/main/scala/org/apache/comet/rules/CometExecRule.scala

@@ -635,15 +635,20 @@ case class CometExecRule(session: SparkSession) extends Rule[SparkPlan] {
         plan
       }
     } else {
-      val normalizedPlan = if (CometConf.COMET_REPLACE_SMJ.get()) {
-        normalizePlan(plan).transformUp { case p =>
+      val normalizedPlan = normalizePlan(plan)
+
+      val planWithJoinRewritten = if (CometConf.COMET_REPLACE_SMJ.get()) {
+        normalizedPlan.transformUp { case p =>
           RewriteJoin.rewrite(p)
         }
       } else {
-        normalizePlan(plan)
+        normalizedPlan
       }
 
-      var newPlan = transform(normalizedPlan)
+      println(normalizedPlan)
+      println(planWithJoinRewritten)
+
+      var newPlan = transform(planWithJoinRewritten)
 
       // if the plan cannot be run fully natively then explain why (when appropriate
       // config is enabled)

spark/src/main/scala/org/apache/comet/rules/RewriteJoin.scala

@@ -20,11 +20,13 @@
 package org.apache.comet.rules
 
 import org.apache.spark.sql.catalyst.optimizer.{BuildLeft, BuildRight, BuildSide, JoinSelectionHelper}
-import org.apache.spark.sql.catalyst.plans.LeftAnti
+import org.apache.spark.sql.catalyst.plans.{LeftAnti, LeftSemi}
 import org.apache.spark.sql.catalyst.plans.logical.Join
 import org.apache.spark.sql.execution.{SortExec, SparkPlan}
 import org.apache.spark.sql.execution.joins.{ShuffledHashJoinExec, SortMergeJoinExec}
 
+import org.apache.comet.CometSparkSessionExtensions.withInfo
+
 /**
  * Adapted from equivalent rule in Apache Gluten.
  *
@@ -65,8 +67,12 @@ object RewriteJoin extends JoinSelectionHelper {
   def rewrite(plan: SparkPlan): SparkPlan = plan match {
     case smj: SortMergeJoinExec =>
       getSmjBuildSide(smj) match {
-        case Some(BuildRight) if smj.joinType == LeftAnti =>
-          // https://github.com/apache/datafusion-comet/issues/457
+        case Some(BuildRight) if smj.joinType == LeftAnti || smj.joinType == LeftSemi =>
+          withInfo(
+            smj,
+            "Cannot rewrite SortMergeJoin to HashJoin: BuildRight with LeftSemi/LeftAnti is not supported")
+          // LeftAnti https://github.com/apache/datafusion-comet/issues/457
+          // LeftSemi https://github.com/apache/datafusion-comet/issues/2667
           plan
         case Some(buildSide) =>
           ShuffledHashJoinExec(