Add in config to avoid AST in some join cases

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

@@ -64,7 +64,11 @@ class GpuShuffledHashJoinMeta(
 
   override def convertToGpu(): GpuExec = {
     val condition = conditionMeta.map(_.convertToGpu())
-    val (joinCondition, filterCondition) = if (conditionMeta.forall(_.canThisBeAst)) {
+    val preferAst = join.joinType match {
+      case _: InnerLike => conf.preferAstJoin
+      case _ => true
+    }
+    val (joinCondition, filterCondition) = if (preferAst && conditionMeta.forall(_.canThisBeAst)) {
       (condition, None)
     } else {
       (None, condition)

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

@@ -16,7 +16,7 @@
 
 package com.nvidia.spark.rapids
 
-import org.apache.spark.sql.catalyst.plans.{FullOuter, Inner}
+import org.apache.spark.sql.catalyst.plans.{FullOuter, Inner, InnerLike}
 import org.apache.spark.sql.execution.SortExec
 import org.apache.spark.sql.execution.joins.SortMergeJoinExec
 import org.apache.spark.sql.rapids.execution.{GpuHashJoin, JoinTypeChecks}
@@ -76,7 +76,11 @@ class GpuSortMergeJoinMeta(
 
   override def convertToGpu(): GpuExec = {
     val condition = conditionMeta.map(_.convertToGpu())
-    val (joinCondition, filterCondition) = if (conditionMeta.forall(_.canThisBeAst)) {
+    val preferAst = join.joinType match {
+      case _: InnerLike => conf.preferAstJoin
+      case _ => true
+    }
+    val (joinCondition, filterCondition) = if (preferAst && conditionMeta.forall(_.canThisBeAst)) {
       (condition, None)
     } else {
       (None, condition)

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

@@ -640,6 +640,12 @@ val GPU_COREDUMP_PIPE_PATTERN = conf("spark.rapids.gpu.coreDump.pipePattern")
     .booleanConf
     .createWithDefault(true)
 
+  val PREFER_AST_JOIN = conf("spark.rapids.sql.join.preferAst")
+      .doc("Prefer the use of AST for non-equality portions of a join")
+      .internal()
+      .booleanConf
+      .createWithDefault(true)
+
   val STABLE_SORT = conf("spark.rapids.sql.stableSort.enabled")
       .doc("Enable or disable stable sorting. Apache Spark's sorting is typically a stable " +
           "sort, but sort stability cannot be guaranteed in distributed work loads because the " +
@@ -2334,6 +2340,8 @@ class RapidsConf(conf: Map[String, String]) extends Logging {
 
   lazy val useShuffledSymmetricHashJoin: Boolean = get(USE_SHUFFLED_SYMMETRIC_HASH_JOIN)
 
+  lazy val preferAstJoin: Boolean = get(PREFER_AST_JOIN)
+
   lazy val stableSort: Boolean = get(STABLE_SORT)
 
   lazy val isFileScanPrunePartitionEnabled: Boolean = get(FILE_SCAN_PRUNE_PARTITION_ENABLED)

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

@@ -31,13 +31,18 @@
 {"spark": "332cdh"}
 {"spark": "333"}
 {"spark": "334"}
+{"spark": "340"}
+{"spark": "341"}
+{"spark": "342"}
+{"spark": "350"}
+{"spark": "351"}
 spark-rapids-shim-json-lines ***/
 package org.apache.spark.sql.rapids.execution
 
 import com.nvidia.spark.rapids._
 
 import org.apache.spark.sql.catalyst.expressions.Expression
-import org.apache.spark.sql.catalyst.plans.JoinType
+import org.apache.spark.sql.catalyst.plans.{InnerLike, JoinType}
 import org.apache.spark.sql.execution.SparkPlan
 import org.apache.spark.sql.execution.joins.BroadcastHashJoinExec
 
@@ -49,7 +54,11 @@ class GpuBroadcastHashJoinMeta(
 
   override def convertToGpu(): GpuExec = {
     val condition = conditionMeta.map(_.convertToGpu())
-    val (joinCondition, filterCondition) = if (conditionMeta.forall(_.canThisBeAst)) {
+    val preferAst = join.joinType match {
+      case _: InnerLike => conf.preferAstJoin
+      case _ => true
+    }
+    val (joinCondition, filterCondition) = if (preferAst && conditionMeta.forall(_.canThisBeAst)) {
       (condition, None)
     } else {
       (None, condition)
@@ -82,4 +91,4 @@ case class GpuBroadcastHashJoinExec(
     override val condition: Option[Expression],
     left: SparkPlan,
     right: SparkPlan) extends GpuBroadcastHashJoinExecBase(
-      leftKeys, rightKeys, joinType, buildSide, condition, left, right)
+      leftKeys, rightKeys, joinType, buildSide, condition, left, right)

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

@@ -29,7 +29,7 @@ import org.apache.spark.TaskContext
 import org.apache.spark.rapids.shims.GpuShuffleExchangeExec
 import org.apache.spark.rdd.RDD
 import org.apache.spark.sql.catalyst.expressions.{Attribute, Expression}
-import org.apache.spark.sql.catalyst.plans.JoinType
+import org.apache.spark.sql.catalyst.plans.{InnerLike, JoinType}
 import org.apache.spark.sql.catalyst.plans.physical.{BroadcastDistribution, Distribution, UnspecifiedDistribution}
 import org.apache.spark.sql.execution.{CoalescedPartitionSpec, SparkPlan}
 import org.apache.spark.sql.execution.adaptive.ShuffleQueryStageExec
@@ -46,7 +46,11 @@ class GpuBroadcastHashJoinMeta(
 
   override def convertToGpu(): GpuExec = {
     val condition = conditionMeta.map(_.convertToGpu())
-    val (joinCondition, filterCondition) = if (conditionMeta.forall(_.canThisBeAst)) {
+    val preferAst = join.joinType match {
+      case _: InnerLike => conf.preferAstJoin
+      case _ => true
+    }
+    val (joinCondition, filterCondition) = if (preferAst && conditionMeta.forall(_.canThisBeAst)) {
       (condition, None)
     } else {
       (None, condition)