Improve performance of BloomFilterMightContain

[andygrove]

What is the problem the feature request solves?

Comet currently performances poorly with the following query when broadcast hash joins are disabled and when Comet native shuffle is disabled.

select ss_sold_date_sk, ss_sold_time_sk, ss_quantity, d_year, d_moy, d_dom
from date_dim join store_sales on d_date_sk = ss_sold_date_sk
where d_year = 2000;

Benchmark results (running aginst sf=100 dataset):

AMD Ryzen 9 7950X3D 16-Core Processor
TPCDS Micro Benchmarks:                   Best Time(ms)   Avg Time(ms)   Stdev(ms)    Rate(M/s)   Per Row(ns)   Relative
------------------------------------------------------------------------------------------------------------------------
join_inner                                          512            537          35        562.4           1.8       1.0X
join_inner: Comet (Scan)                            738            750          13        390.3           2.6       0.7X
join_inner: Comet (Scan, Exec)                     4564           4724         226         63.1          15.8       0.1X

Sample native plan with metrics:

FilterExec: col_0@0 IS NOT NULL AND BloomFilterMightContain [bloom_filter_expr: Subquery [id: 2851], value_expr: xxhash64(col_0@0, 42)], metrics=[output_rows=605440, elapsed_compute=85.205965ms]
  ScanExec: source=[CometScan parquet spark_catalog.default.store_sales (unknown)], schema=[col_0: Int32, col_1: Int32, col_2: Int32], metrics=[output_rows=3145728, elapsed_compute=19.12706ms]

Describe the potential solution

No response

Additional context

No response

[eejbyfeldt]

EDIT: I must have messed something up. Looking at the queryplan my test run did not trigger the BloomMightContain logic. I guss I missed this part broadcast hash joins are disabled in the ticket.

I had a quick look at this in a profiler and to me it did not look like much time was spent in the BloomFilterMightContain related code. The things that stood out as taking significant time was copying of data due to unpacking of dictionaries during the scan operation here: https://github.com/eejbyfeldt/datafusion-comet/blob/a99f7428398793507b31188c8919e4cf128d8d38/native/core/src/execution/operators/scan.rs#L353-L370 and the copying done in the FilterExec which sounds similar to what is discussed in https://github.com/apache/datafusion-comet/issues/808

So maybe this ticket should be replace with one about removing the unpacking of dictionaries in the scan operator.