Support existence join type for broadcast nested loop join #5301

res-life · 2022-04-22T07:14:58Z

Closes #5034
Support existence join type for broadcast nested loop join

Signed-off-by: Chong Gao res_life@163.com

Signed-off-by: Chong Gao <res_life@163.com>

res-life · 2022-04-22T07:15:44Z

build

res-life · 2022-04-22T07:21:36Z

JoinGatherer for existence join
Existence join generates an exists boolean column with true or false in it,
then appends it to the output columns. It does not shrink or expand left table.
Just provides the exists column for the next following operator(usually filter).

e.g.:
select * from left_table where
  left_table.column_0 >= 3
  or
  exists (select * from right_table where left_table.column_1 < right_table.column_1)

Explanation of this sql is:

Filter(left_table.column_0 >= 3 or `exists`)
  Existence_join // generate `exists` column, do not shrink or expand the rows of left table
    left_table
    right_table

This is to add a new JoinGatherer to handle this existence join.

firestarman

Took a quick review, and got some comments.

firestarman · 2022-04-22T07:25:23Z

integration_tests/src/main/python/join_test.py

+
+
+@ignore_order
+@pytest.mark.parametrize('aqeEnabled', [pytest.param(True, id='aqe:on'), pytest.param(False, id='aqe:off')])


Suggested change

@pytest.mark.parametrize('aqeEnabled', [pytest.param(True, id='aqe:on'), pytest.param(False, id='aqe:off')])

@pytest.mark.parametrize('aqeEnabled', [True, False], ids=['AQE_ON', 'AQE_OFF'])

We should have a consistent id scheme across the pytest code base regardless of the choice made

firestarman · 2022-04-22T07:27:28Z

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

-      outOfBoundsPolicy: OutOfBoundsPolicy): JoinGatherer =
-    new JoinGathererImpl(gatherMap, inputData, outOfBoundsPolicy)
+      outOfBoundsPolicy: OutOfBoundsPolicy, isExistenceJoin: Boolean = false): JoinGatherer = {
+    if (!isExistenceJoin) {


NIT:

if (isExistenceJoin) { new JoinGathererForExistenceJoin(gatherMap, inputData, outOfBoundsPolicy) } else { new JoinGathererImpl(gatherMap, inputData, outOfBoundsPolicy) }

firestarman · 2022-04-22T07:58:16Z

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

+ * </code>
+ */
+class JoinGathererForExistenceJoin(


NIT

Suggested change

class JoinGathererForExistenceJoin(

class ExistenceJoinGathererImpl(

No need now.

firestarman · 2022-04-22T08:17:26Z

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

+          // `exists.numRows` == `batch.numRows`,
+          //  with true or false in it indicating if the row is gathered
+          withResource(genExistsColumn(gatherView, batch.numRows())) { exists =>


Personally I perfer to generate the column of existence according to the subTableCbTmp, not the whole batch. Then there is no need to split it next. Instead you can append the generated 'existence' column to the sub table directly.

Just as @firestarman mentioned, this may lead to a lot of duplicated runs of full-scale scatter. IIUC, if we split the gather map as left table, we need to transform the global offset to the local offset before scattering. Perhaps there is another approach, just cache the exist column for reuse over mini batches.

Emit the batch directly as the exists boolean column is relatively small.
It's the same as GpuHashJoin does, no need to split now.

Signed-off-by: Chong Gao <res_life@163.com>

res-life · 2022-04-22T13:24:13Z

build

integration_tests/src/main/python/join_test.py

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

res-life · 2022-04-24T07:42:22Z

build

jlowe

Minor nits on the class names used but otherwise looking good.

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

...in/src/main/scala/org/apache/spark/sql/rapids/execution/GpuBroadcastNestedLoopJoinExec.scala

Signed-off-by: Chong Gao <res_life@163.com>

res-life · 2022-04-26T02:15:19Z

build

gerashegalov

LGTM, some nits

gerashegalov · 2022-04-26T22:04:05Z

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

@@ -601,7 +572,7 @@ class ExistenceJoinIterator(
   * the value "false", scattering "true" into column FC will produce the "exists"
   * column of ExistenceJoin
   */
-  private def existsScatterMap(leftColumnarBatch: ColumnarBatch): GatherMap = {
+  def existsScatterMap(leftColumnarBatch: ColumnarBatch): GatherMap = {


nit:

Suggested change

def existsScatterMap(leftColumnarBatch: ColumnarBatch): GatherMap = {

override def existsScatterMap(leftColumnarBatch: ColumnarBatch): GatherMap = {

gerashegalov · 2022-04-26T22:05:06Z

...in/src/main/scala/org/apache/spark/sql/rapids/execution/GpuBroadcastNestedLoopJoinExec.scala

+
+  use(condition)
+
+  def existsScatterMap(leftColumnarBatch: ColumnarBatch): GatherMap = {


nit:

Suggested change

def existsScatterMap(leftColumnarBatch: ColumnarBatch): GatherMap = {

override def existsScatterMap(leftColumnarBatch: ColumnarBatch): GatherMap = {

gerashegalov · 2022-04-26T22:05:06Z

...in/src/main/scala/org/apache/spark/sql/rapids/execution/GpuBroadcastNestedLoopJoinExec.scala

+
+  use(condition)
+
+  def existsScatterMap(leftColumnarBatch: ColumnarBatch): GatherMap = {


nit:

Suggested change

def existsScatterMap(leftColumnarBatch: ColumnarBatch): GatherMap = {

override def existsScatterMap(leftColumnarBatch: ColumnarBatch): GatherMap = {

gerashegalov · 2022-04-26T22:11:21Z

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

@@ -601,7 +572,7 @@ class ExistenceJoinIterator(
   * the value "false", scattering "true" into column FC will produce the "exists"
   * column of ExistenceJoin
   */


the comment should be moved to the abstract parent class method declaration

gerashegalov · 2022-04-26T22:11:21Z

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

@@ -601,7 +572,7 @@ class ExistenceJoinIterator(
   * the value "false", scattering "true" into column FC will produce the "exists"
   * column of ExistenceJoin
   */


the comment should be moved to the abstract parent class method declaration

res-life · 2022-04-27T01:16:16Z

build

res-life · 2022-04-27T03:21:41Z

build

gerashegalov · 2022-04-26T22:04:05Z

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

@@ -601,7 +572,7 @@ class ExistenceJoinIterator(
   * the value "false", scattering "true" into column FC will produce the "exists"
   * column of ExistenceJoin
   */
-  private def existsScatterMap(leftColumnarBatch: ColumnarBatch): GatherMap = {
+  def existsScatterMap(leftColumnarBatch: ColumnarBatch): GatherMap = {


nit:

Suggested change

def existsScatterMap(leftColumnarBatch: ColumnarBatch): GatherMap = {

override def existsScatterMap(leftColumnarBatch: ColumnarBatch): GatherMap = {

Support existence join type for broadcast nested loop join

f2a24df

Signed-off-by: Chong Gao <res_life@163.com>

res-life requested review from jlowe, gerashegalov and firestarman April 22, 2022 07:15

firestarman reviewed Apr 22, 2022

View reviewed changes

Refactor

99de6c9

Signed-off-by: Chong Gao <res_life@163.com>

jlowe reviewed Apr 22, 2022

View reviewed changes

Update

b138837

sameerz added the feature request New feature or request label Apr 25, 2022

sameerz added this to the Apr 18 - Apr 29 milestone Apr 25, 2022

jlowe reviewed Apr 25, 2022

View reviewed changes

sql-plugin/src/main/scala/org/apache/spark/sql/rapids/execution/GpuHashJoin.scala Outdated Show resolved Hide resolved

...in/src/main/scala/org/apache/spark/sql/rapids/execution/GpuBroadcastNestedLoopJoinExec.scala Outdated Show resolved Hide resolved

Chong Gao added 2 commits April 26, 2022 09:16

Merge branch 'branch-22.06' into existence-join

beff080

Rename class name

b7ecb2f

Signed-off-by: Chong Gao <res_life@163.com>

jlowe previously approved these changes Apr 26, 2022

View reviewed changes

gerashegalov reviewed Apr 26, 2022

View reviewed changes

Fix nits

60082bb

res-life dismissed jlowe’s stale review via 60082bb April 27, 2022 01:15

Fix a nit

8eb5d64

gerashegalov approved these changes Apr 27, 2022

View reviewed changes

res-life merged commit 45d6fcc into NVIDIA:branch-22.06 Apr 27, 2022

res-life deleted the existence-join branch April 27, 2022 11:34

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Support existence join type for broadcast nested loop join #5301

Support existence join type for broadcast nested loop join #5301

res-life commented Apr 22, 2022 •

edited

Loading

res-life commented Apr 22, 2022

res-life commented Apr 22, 2022

firestarman left a comment

firestarman Apr 22, 2022

gerashegalov Apr 26, 2022

res-life Apr 27, 2022

firestarman Apr 22, 2022

res-life Apr 27, 2022

firestarman Apr 22, 2022

res-life Apr 27, 2022

firestarman Apr 22, 2022

sperlingxx Apr 22, 2022

res-life Apr 27, 2022

res-life commented Apr 22, 2022

res-life commented Apr 24, 2022

jlowe left a comment

res-life commented Apr 26, 2022

gerashegalov left a comment

gerashegalov Apr 26, 2022

res-life Apr 27, 2022

gerashegalov Apr 26, 2022

res-life Apr 27, 2022

gerashegalov Apr 26, 2022

res-life Apr 27, 2022

gerashegalov Apr 26, 2022

res-life Apr 27, 2022

gerashegalov Apr 26, 2022

res-life Apr 27, 2022

res-life commented Apr 27, 2022

res-life commented Apr 27, 2022

gerashegalov Apr 26, 2022



		@ignore_order
		@pytest.mark.parametrize('aqeEnabled', [pytest.param(True, id='aqe:on'), pytest.param(False, id='aqe:off')])

	class JoinGathererForExistenceJoin(
	class ExistenceJoinGathererImpl(

	def existsScatterMap(leftColumnarBatch: ColumnarBatch): GatherMap = {
	override def existsScatterMap(leftColumnarBatch: ColumnarBatch): GatherMap = {


		use(condition)

		def existsScatterMap(leftColumnarBatch: ColumnarBatch): GatherMap = {

Support existence join type for broadcast nested loop join #5301

Support existence join type for broadcast nested loop join #5301

Conversation

res-life commented Apr 22, 2022 • edited Loading

res-life commented Apr 22, 2022

res-life commented Apr 22, 2022

firestarman left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

res-life commented Apr 22, 2022

res-life commented Apr 24, 2022

jlowe left a comment

Choose a reason for hiding this comment

res-life commented Apr 26, 2022

gerashegalov left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

res-life commented Apr 27, 2022

res-life commented Apr 27, 2022

Choose a reason for hiding this comment

res-life commented Apr 22, 2022 •

edited

Loading