Keep post-filter coalesce for hash joins

NVIDIA · Oct 5, 2021 · cb20aee · cb20aee
1 parent fcfb436
commit cb20aee
Show file tree

Hide file tree

Showing 11 changed files with 16 additions and 30 deletions.
diff --git a/...ugin/src/main/301db/scala/com/nvidia/spark/rapids/shims/v2/GpuBroadcastHashJoinExec.scala b/...ugin/src/main/301db/scala/com/nvidia/spark/rapids/shims/v2/GpuBroadcastHashJoinExec.scala
@@ -83,9 +83,7 @@ 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, coalesceAfter = false)
-    }.getOrElse(joinExec)
+    condition.map(c => GpuFilterExec(c.convertToGpu(), joinExec)).getOrElse(joinExec)
   }
 }
 

diff --git a/...lugin/src/main/301db/scala/com/nvidia/spark/rapids/shims/v2/GpuShuffledHashJoinExec.scala b/...lugin/src/main/301db/scala/com/nvidia/spark/rapids/shims/v2/GpuShuffledHashJoinExec.scala
@@ -72,9 +72,7 @@ class GpuShuffledHashJoinMeta(
       join.rightKeys)
     // 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, coalesceAfter = false)
-    }.getOrElse(joinExec)
+    condition.map(c => GpuFilterExec(c.convertToGpu(), joinExec)).getOrElse(joinExec)
   }
 }
 

diff --git a/sql-plugin/src/main/301db/scala/com/nvidia/spark/rapids/shims/v2/GpuSortMergeJoinMeta.scala b/sql-plugin/src/main/301db/scala/com/nvidia/spark/rapids/shims/v2/GpuSortMergeJoinMeta.scala
@@ -91,9 +91,7 @@ class GpuSortMergeJoinMeta(
       join.rightKeys)
     // 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, coalesceAfter = false)
-    }.getOrElse(joinExec)
+    condition.map(c => GpuFilterExec(c.convertToGpu(), joinExec)).getOrElse(joinExec)
   }
 
   /**

diff --git a/...n/301until310-nondb/scala/com/nvidia/spark/rapids/shims/v2/GpuBroadcastHashJoinExec.scala b/...n/301until310-nondb/scala/com/nvidia/spark/rapids/shims/v2/GpuBroadcastHashJoinExec.scala
@@ -83,9 +83,7 @@ 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, coalesceAfter = false)
-    }.getOrElse(joinExec)
+    condition.map(c => GpuFilterExec(c.convertToGpu(), joinExec)).getOrElse(joinExec)
   }
 }
 

diff --git a/...in/301until310-nondb/scala/com/nvidia/spark/rapids/shims/v2/GpuShuffledHashJoinExec.scala b/...in/301until310-nondb/scala/com/nvidia/spark/rapids/shims/v2/GpuShuffledHashJoinExec.scala
@@ -71,9 +71,7 @@ class GpuShuffledHashJoinMeta(
       join.rightKeys)
     // 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, coalesceAfter = false)
-    }.getOrElse(joinExec)
+    condition.map(c => GpuFilterExec(c.convertToGpu(), joinExec)).getOrElse(joinExec)
   }
 }
 

diff --git a/.../main/301until310-nondb/scala/com/nvidia/spark/rapids/shims/v2/GpuSortMergeJoinMeta.scala b/.../main/301until310-nondb/scala/com/nvidia/spark/rapids/shims/v2/GpuSortMergeJoinMeta.scala
@@ -90,9 +90,7 @@ class GpuSortMergeJoinMeta(
       join.rightKeys)
     // 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, coalesceAfter = false)
-    }.getOrElse(joinExec)
+    condition.map(c => GpuFilterExec(c.convertToGpu(), joinExec)).getOrElse(joinExec)
   }
 
   /**

diff --git a/...in/src/main/311+-all/scala/com/nvidia/spark/rapids/shims/v2/GpuShuffledHashJoinExec.scala b/...in/src/main/311+-all/scala/com/nvidia/spark/rapids/shims/v2/GpuShuffledHashJoinExec.scala
@@ -72,9 +72,7 @@ class GpuShuffledHashJoinMeta(
       join.rightKeys)
     // 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, coalesceAfter = false)
-    }.getOrElse(joinExec)
+    condition.map(c => GpuFilterExec(c.convertToGpu(), joinExec)).getOrElse(joinExec)
   }
 }
 

diff --git a/...lugin/src/main/311+-all/scala/com/nvidia/spark/rapids/shims/v2/GpuSortMergeJoinMeta.scala b/...lugin/src/main/311+-all/scala/com/nvidia/spark/rapids/shims/v2/GpuSortMergeJoinMeta.scala
@@ -91,9 +91,7 @@ class GpuSortMergeJoinMeta(
       join.rightKeys)
     // 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, coalesceAfter = false)
-    }.getOrElse(joinExec)
+    condition.map(c => GpuFilterExec(c.convertToGpu(), joinExec)).getOrElse(joinExec)
   }
 
   /**

diff --git a/...src/main/311+-nondb/scala/com/nvidia/spark/rapids/shims/v2/GpuBroadcastHashJoinExec.scala b/...src/main/311+-nondb/scala/com/nvidia/spark/rapids/shims/v2/GpuBroadcastHashJoinExec.scala
@@ -85,9 +85,7 @@ 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, coalesceAfter = false)
-    }.getOrElse(joinExec)
+    condition.map(c => GpuFilterExec(c.convertToGpu(), joinExec)).getOrElse(joinExec)
   }
 }
 

diff --git a/...ugin/src/main/311db/scala/com/nvidia/spark/rapids/shims/v2/GpuBroadcastHashJoinExec.scala b/...ugin/src/main/311db/scala/com/nvidia/spark/rapids/shims/v2/GpuBroadcastHashJoinExec.scala
@@ -83,9 +83,7 @@ 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, coalesceAfter = false)
-    }.getOrElse(joinExec)
+    condition.map(c => GpuFilterExec(c.convertToGpu(), joinExec)).getOrElse(joinExec)
   }
 }
 

diff --git a/...main/scala/org/apache/spark/sql/rapids/execution/GpuBroadcastNestedLoopJoinExecBase.scala b/...main/scala/org/apache/spark/sql/rapids/execution/GpuBroadcastNestedLoopJoinExecBase.scala
@@ -122,6 +122,12 @@ class GpuBroadcastNestedLoopJoinMeta(
     } else {
       // condition cannot be implemented via AST so fallback to a post-filter if necessary
       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)
     }