Avoid using AST on inner joins and avoid coalesce after nested loop join filter

sql-plugin/src/main/301db/scala/com/nvidia/spark/rapids/shims/v2/GpuBroadcastHashJoinExec.scala

@@ -83,7 +83,9 @@ class GpuBroadcastHashJoinMeta(
       left, right)
     // The GPU does not yet support conditional joins, so conditions are implemented
     // as a filter after the join when possible.
-    condition.map(c => GpuFilterExec(c.convertToGpu(), joinExec)).getOrElse(joinExec)
+    condition.map {
+      c => GpuFilterExec(c.convertToGpu(), joinExec, coalesceAfter = false)
+    }.getOrElse(joinExec)
   }
 }
 

sql-plugin/src/main/301db/scala/com/nvidia/spark/rapids/shims/v2/GpuShuffledHashJoinExec.scala

@@ -70,7 +70,9 @@ class GpuShuffledHashJoinMeta(
       isSkewJoin = false)
     // The GPU does not yet support conditional joins, so conditions are implemented
     // as a filter after the join when possible.
-    condition.map(c => GpuFilterExec(c.convertToGpu(), joinExec)).getOrElse(joinExec)
+    condition.map {
+      c => GpuFilterExec(c.convertToGpu(), joinExec, coalesceAfter = false)
+    }.getOrElse(joinExec)
   }
 }
 

sql-plugin/src/main/301db/scala/com/nvidia/spark/rapids/shims/v2/GpuSortMergeJoinExec.scala

@@ -89,7 +89,9 @@ class GpuSortMergeJoinMeta(
       join.isSkewJoin)
     // The GPU does not yet support conditional joins, so conditions are implemented
     // as a filter after the join when possible.
-    condition.map(c => GpuFilterExec(c.convertToGpu(), joinExec)).getOrElse(joinExec)
+    condition.map {
+      c => GpuFilterExec(c.convertToGpu(), joinExec, coalesceAfter = false)
+    }.getOrElse(joinExec)
   }
 
   /**

sql-plugin/src/main/301until310-nondb/scala/com/nvidia/spark/rapids/shims/v2/GpuBroadcastHashJoinExec.scala

@@ -83,7 +83,9 @@ class GpuBroadcastHashJoinMeta(
       left, right)
     // The GPU does not yet support conditional joins, so conditions are implemented
     // as a filter after the join when possible.
-    condition.map(c => GpuFilterExec(c.convertToGpu(), joinExec)).getOrElse(joinExec)
+    condition.map { c =>
+      GpuFilterExec(c.convertToGpu(), joinExec, coalesceAfter = false)
+    }.getOrElse(joinExec)
   }
 }
 

sql-plugin/src/main/301until310-nondb/scala/com/nvidia/spark/rapids/shims/v2/GpuShuffledHashJoinExec.scala

@@ -69,7 +69,9 @@ class GpuShuffledHashJoinMeta(
       isSkewJoin = false)
     // The GPU does not yet support conditional joins, so conditions are implemented
     // as a filter after the join when possible.
-    condition.map(c => GpuFilterExec(c.convertToGpu(), joinExec)).getOrElse(joinExec)
+    condition.map {
+      c => GpuFilterExec(c.convertToGpu(), joinExec, coalesceAfter = false)
+    }.getOrElse(joinExec)
   }
 }
 

sql-plugin/src/main/301until310-nondb/scala/com/nvidia/spark/rapids/shims/v2/GpuSortMergeJoinMeta.scala

@@ -88,7 +88,9 @@ class GpuSortMergeJoinMeta(
       join.isSkewJoin)
     // The GPU does not yet support conditional joins, so conditions are implemented
     // as a filter after the join when possible.
-    condition.map(c => GpuFilterExec(c.convertToGpu(), joinExec)).getOrElse(joinExec)
+    condition.map {
+      c => GpuFilterExec(c.convertToGpu(), joinExec, coalesceAfter = false)
+    }.getOrElse(joinExec)
   }
 
   /**

sql-plugin/src/main/311+-all/scala/com/nvidia/spark/rapids/shims/v2/GpuShuffledHashJoinExec.scala

@@ -70,7 +70,9 @@ class GpuShuffledHashJoinMeta(
       isSkewJoin = false)
     // The GPU does not yet support conditional joins, so conditions are implemented
     // as a filter after the join when possible.
-    condition.map(c => GpuFilterExec(c.convertToGpu(), joinExec)).getOrElse(joinExec)
+    condition.map {
+      c => GpuFilterExec(c.convertToGpu(), joinExec, coalesceAfter = false)
+    }.getOrElse(joinExec)
   }
 }
 

sql-plugin/src/main/311+-all/scala/com/nvidia/spark/rapids/shims/v2/GpuSortMergeJoinExec.scala

@@ -89,7 +89,9 @@ class GpuSortMergeJoinMeta(
       join.isSkewJoin)
     // The GPU does not yet support conditional joins, so conditions are implemented
     // as a filter after the join when possible.
-    condition.map(c => GpuFilterExec(c.convertToGpu(), joinExec)).getOrElse(joinExec)
+    condition.map {
+      c => GpuFilterExec(c.convertToGpu(), joinExec, coalesceAfter = false)
+    }.getOrElse(joinExec)
   }
 
   /**

sql-plugin/src/main/311+-nondb/scala/com/nvidia/spark/rapids/shims/v2/GpuBroadcastHashJoinExec.scala

@@ -85,7 +85,9 @@ class GpuBroadcastHashJoinMeta(
       left, right)
     // The GPU does not yet support conditional joins, so conditions are implemented
     // as a filter after the join when possible.
-    condition.map(c => GpuFilterExec(c.convertToGpu(), joinExec)).getOrElse(joinExec)
+    condition.map {
+      c => GpuFilterExec(c.convertToGpu(), joinExec, coalesceAfter = false)
+    }.getOrElse(joinExec)
   }
 }
 

sql-plugin/src/main/311db/scala/com/nvidia/spark/rapids/shims/v2/GpuBroadcastHashJoinExec.scala

@@ -83,7 +83,9 @@ class GpuBroadcastHashJoinMeta(
       left, right)
     // The GPU does not yet support conditional joins, so conditions are implemented
     // as a filter after the join when possible.
-    condition.map(c => GpuFilterExec(c.convertToGpu(), joinExec)).getOrElse(joinExec)
+    condition.map {
+      c => GpuFilterExec(c.convertToGpu(), joinExec, coalesceAfter = false)
+    }.getOrElse(joinExec)
   }
 }
 

sql-plugin/src/main/scala/com/nvidia/spark/rapids/basicPhysicalOperators.scala

@@ -310,7 +310,10 @@ object GpuFilter extends Arm {
   }
 }
 
-case class GpuFilterExec(condition: Expression, child: SparkPlan)
+case class GpuFilterExec(
+    condition: Expression,
+    child: SparkPlan,
+    override val coalesceAfter: Boolean = true)
     extends ShimUnaryExecNode with GpuPredicateHelper with GpuExec {
 
   override lazy val additionalMetrics: Map[String, GpuMetric] = Map(
@@ -322,11 +325,6 @@ case class GpuFilterExec(condition: Expression, child: SparkPlan)
     case _ => false
   }
 
-  /**
-   * Potentially a lot is removed.
-   */
-  override def coalesceAfter: Boolean = true
-
   // If one expression and its children are null intolerant, it is null intolerant.
   private def isNullIntolerant(expr: Expression): Boolean = expr match {
     case e: NullIntolerant => e.children.forall(isNullIntolerant)

sql-plugin/src/main/scala/org/apache/spark/sql/rapids/execution/GpuBroadcastNestedLoopJoinExecBase.scala

@@ -93,7 +93,13 @@ class GpuBroadcastNestedLoopJoinMeta(
     verifyBuildSideWasReplaced(buildSide)
 
     val condition = conditionMeta.map(_.convertToGpu())
-    val isAstCondition = conditionMeta.forall(_.canThisBeAst)
+    val isAstCondition = join.joinType match {
+      case _: InnerLike =>
+        // It appears to be faster to manifest the full cross join and post-filter than
+        // evaluate the AST during the join.
+        false
+      case _ => conditionMeta.forall(_.canThisBeAst)
+    }
     join.joinType match {
       case _: InnerLike =>
       case LeftOuter | LeftSemi | LeftAnti if gpuBuildSide == GpuBuildLeft =>
@@ -115,7 +121,9 @@ class GpuBroadcastNestedLoopJoinMeta(
       joinExec
     } else {
       // condition cannot be implemented via AST so fallback to a post-filter if necessary
-      condition.map(c => GpuFilterExec(c, joinExec)).getOrElse(joinExec)
+      condition.map {
+        c => GpuFilterExec(c, joinExec, coalesceAfter = false)
+      }.getOrElse(joinExec)
     }
   }
 }