[FEA] Support ExistenceJoin

[revans2]

Is your feature request related to a problem? Please describe.
Spark will some times use an optimized join type called an ExistenceJoin. It is not a direct join type you can just do but instead an optimized version of the IN or EXISTS operators that lets spark remove a subquery. This shows up in TPC-DS query 10, and probably others. It is very similar to a left semi join, but instead of filtering out columns that don't match instead it adds a new column with a boolean true or false to indicate if the join would have matched.

Describe the solution you'd like
I think we might need some support from cudf for this.

Describe alternatives you've considered
There isn't a lot here. I think we need some help from cudf.

• add an integration test exhibiting ExistenceJoin Add preliminary test and test framework changes for ExistanceJoin #4697
• implement a gatherer for ExistenceJoin Implement ExistenceJoin Iterator using an auxiliary left semijoin  #4796

[viadea]

mini repro:

select
  count(*)
from
  customer c
where
  exists (select * from store_sales
          where c.c_customer_sk = ss_customer_sk) 
          AND
   (exists (select * from web_sales
            where c.c_customer_sk = ws_bill_customer_sk) or
    exists (select * from catalog_sales
            where c.c_customer_sk = cs_ship_customer_sk))

Driver Log message:

scala> spark.sql(query).explain()
21/11/03 23:19:56 WARN GpuOverrides:
          !Exec <SortMergeJoinExec> cannot run on GPU because ExistenceJoin(exists#225) currently is not supported
            @Expression <AttributeReference> c_customer_sk#0 could run on GPU
            @Expression <AttributeReference> cs_ship_customer_sk#156 could run on GPU
            !Exec <SortMergeJoinExec> cannot run on GPU because ExistenceJoin(exists#224) currently is not supported
              @Expression <AttributeReference> c_customer_sk#0 could run on GPU
              @Expression <AttributeReference> ws_bill_customer_sk#85 could run on GPU

[revans2]

We might be able to do this without any changes from CUDF. ExistenceJoin is not a regular join type, as was stated before. As such cudf might push back a bit if we try to implement it there. Happily ExistanceJoin really just produces the left output columns, unchanged, and adds in an "exists" boolean column. The exists boolean columns is the same thing that is used by Spark to filter the left table on a semi-join.

As such we can produce the gather map for a left semi-join, and instead of gathering the results, we can look to see if the column would have been gathered.

I agree that in general if we can get left-semi or left-anti to optionally spit out a boolean column instead of a gather map, that would be ideal, but we should be able to do it without any help if needed.

[jlowe]

I think we can generate the existence column from the semi-join gather map like this:

1. Create a boolean ColumnVector of FALSE values with as many rows as the left table
2. Build a Table with that column vector
3. Use Table.scatter to scatter a TRUE boolean scalar to the above table using the semi join gather map as the scatter map