Implement dynamic discrete pruning through a join

[gruuya]

Is your feature request related to a problem or challenge?

Whilst we currently do have bounds (min/max pruning) pushed down dynamically through a hash join, we don't have the more advanced discrete version of it wired up, even though all the primitives are in place.

Consequently, we're leaving some performance benefits on the table, especially when a qualifier on the build side filters some non-contiguous narrow sub-ranges in a broader contiguous range (e.g. every March in a 5-year interval).

Describe the solution you'd like

Anything that would bring down the scanned row count in the following toy example in line with the build side:

> select version();
+--------------------------------------------+
| version()                                  |
+--------------------------------------------+
| Apache DataFusion 55.0.0, aarch64 on macos |
+--------------------------------------------+
1 row(s) fetched.
Elapsed 0.002 seconds.

> copy (select i as k, random() as v from generate_series(0, 1999999) t(i))
to '/tmp/fact.parquet'
stored as parquet options ('format.max_row_group_size' '1000');
+---------+
| count   |
+---------+
| 2000000 |
+---------+
1 row(s) fetched.
Elapsed 0.093 seconds.

> create external table fact stored as parquet location '/tmp/fact.parquet';
0 row(s) fetched.
Elapsed 0.010 seconds.

> create table dim as
select i as k from generate_series(0, 1999999) t(i) where i % 10000 < 200;
0 row(s) fetched.
Elapsed 0.014 seconds.

> explain analyze select count(*), sum(v) from fact join dim on fact.k = dim.k;
...
|                   |           DataSourceExec: file_groups={12 groups: [[tmp/fact.parquet:0..1929817], [tmp/fact.parquet:1929817..3859634], [tmp/fact.parquet:3859634..5789451], [tmp/fact.parquet:5789451..7719268], [tmp/fact.parquet:7719268..9649085], ...]}, projection=[k, v], output_ordering=[k@0 ASC NULLS LAST], file_type=parquet, predicate=DynamicFilter [ k@0 >= 0 AND k@0 <= 1990199 AND hash_lookup ], dynamic_rg_pruning=eligible, pruning_predicate=k_null_count@1 != row_count@2 AND k_max@0 >= 0 AND k_null_count@1 != row_count@2 AND k_min@3 <= 1990199, required_guarantees=[], metrics=[output_rows=1.99 M, elapsed_compute=584.15µs, output_bytes=248.9 MB, output_batches=1.99 K, files_ranges_pruned_statistics=12 total → 12 matched, row_groups_pruned_statistics=2.00 K total → 1.99 K matched, row_groups_pruned_bloom_filter=1.99 K total → 1.99 K matched, page_index_pages_pruned=1.99 K total → 1.99 K matched, page_index_rows_pruned=1.99 M total → 1.99 M matched, limit_pruned_row_groups=0 total → 0 matched, batches_split=0, bytes_scanned=22.34 M, file_open_errors=0, file_scan_errors=0, files_opened=12, files_processed=12, num_predicate_creation_errors=0, predicate_evaluation_errors=0, pushdown_rows_matched=0, pushdown_rows_pruned=0, row_groups_pruned_dynamic_filter=0, predicate_cache_inner_records=0, predicate_cache_records=0, bloom_filter_eval_time=23.02ms, metadata_load_time=1.30ms, page_index_eval_time=7.02ms, row_pushdown_eval_time=36ns, statistics_eval_time=842.73µs, time_elapsed_opening=28.55ms, time_elapsed_processing=120.79ms, time_elapsed_scanning_total=1.19s, time_elapsed_scanning_until_data=9.18ms, output_rows_skew=1.64%, scan_efficiency_ratio=96.46% (22.34 M/23.16 M)] |
...

In particular, note that all 2 million rows are scanned, even though the dim table (build side) has only 40K values (co-located in 10% of fact's row groups) to join on.

Describe alternatives you've considered

One might be tempted to play around with datafusion.execution.parquet.max_in_list_size/datafusion.optimizer.hash_join_inlist_pushdown_max_distinct_values and thus (ab)use InListExpr to achieve this, which seem to guard this path, but that is probably less than ideal for a number of reasons (with pushdown_filters=true this might prove to be counter-productive, EXPLAINs are polluted, etc.).

Additional context

Related to #7955.