Avoid using AST on inner joins and avoid coalesce after nested loop j…

…oin filter (#3746)

* Avoid using AST on inner joins and avoid coalesce after post-join filter

Signed-off-by: Jason Lowe <jlowe@nvidia.com>

* Keep post-filter coalesce for hash joins

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,15 @@ 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 {
+        // TODO: Restore batch coalescing logic here.
+        // Avoid requesting a post-filter-coalesce here, as we've seen poor performance with
+        // the cross join microbenchmark. This is a short-term hack for the benchmark, and
+        // ultimately this should be solved with the resolution of one or more of the following:
+        // https://github.com/NVIDIA/spark-rapids/issues/3749
+        // https://github.com/NVIDIA/spark-rapids/issues/3750
+        c => GpuFilterExec(c, joinExec, coalesceAfter = false)
+      }.getOrElse(joinExec)
     }
   }
 }