gruuya opened a new pull request, #25602:
URL: https://github.com/apache/datafusion/pull/25602

   Map the build side's keys onto a fixed-size bitmap over their range and test 
container min/max against it, so a scan can skip containers whose values fall 
in the gaps between keys.
   
   The bitmap is bounded at 128 KiB regardless of build side size, and one 
whose buckets are all set is discarded, so a contiguous key set costs nothing. 
The key range and distinct count already exist for the bounds predicate, so a 
dense key set is ruled out before allocating.
   
   Integer keys only; other types go unpruned.
   
   ## Which issue does this PR close?
   
   <!--
   We generally require a GitHub issue to be filed for all bug fixes and 
enhancements and this helps us generate change logs for our releases. You can 
link an issue to this PR using the GitHub syntax. For example `Closes #123` 
indicates that this PR will close issue #123.
   -->
   
   - Closes https://github.com/apache/datafusion/issues/25291.
   - Alternative to https://github.com/apache/datafusion/pull/25292.
   
   ## Rationale for this change
   
   <!--
    Why are you proposing this change? If this is already explained clearly in 
the issue then this section is not needed.
    Explaining clearly why changes are proposed helps reviewers understand your 
changes and offer better suggestions for fixes.  
   
   Please explain the problem you are trying to solve in terms of the 
user-visible
   behavior, rather than the implementation.
   
   For example, "The code in `foo.rs` doesn't handle nulls" is a symptom of the
   implementation. "COUNT(DISTINCT) returns wrong results when the column 
contains
   nulls" is the user-visible problem.
   -->
   
   Avoid scanning redundant files/row-groups/pages in the probe side of hash 
joins, based on the values dictated by the build side.
   
   ## What changes are included in this PR?
   
   <!--
   There is no need to duplicate the description in the issue here, but it is 
sometimes worth providing a summary of the individual changes in this PR.
   -->
   
    - wiring up PushdownStrategy::Map/HashTableLookupExpr to carry a 
representation of the build side values from a hash join, through
    - `KeyRangeBitmap` implementation, which maps the build-side values array 
into a finite-sized bitmap bucket, and can answer probing questions for certain 
ranges
    - extend `build_predicate_expression` to build the associated pruning 
expression from `HashTableLookupExpr` using the new 
`KeyRangeBitmapPruningExpr`, which implements `PhysicalExpr` on top of 
`KeyRangeBitmap` 
    - also extend `build_predicate_expression` so that it now pushes down 
pruning for `CaseExprs`, since that also unlocks the partitioned hash-join 
scenario this pr targets
   
   ## What is the testing strategy for this PR?
   
   <!--
   We typically require tests for all PRs in order to:
   1. Prevent the code from being accidentally broken by subsequent changes
   2. Serve as another way to document the expected behavior of the code
   
   Briefly describe how this PR is tested, and point to the specific tests you 
added. For example: 'This new feature is covered by the `sqllogictest` cases 
added in `foo.slt`'.
   
   If this PR does not add tests, explain why. For example, if the change is 
already covered by existing tests, please mention it.
   
   You should also check the `codecov` bot reply on this PR to confirm the 
changed code is exercised.
   -->
   
   A number of unit tests added, and one SLT added.
   
   Also verified on the query shape that motivated the original issue
   ```sql
   > 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.008 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.
   
   > select count(*), sum(v) from fact join dim on fact.k = dim.k;
   select count(*), sum(v) from fact join dim on fact.k = dim.k;
   select count(*), sum(v) from fact join dim on fact.k = dim.k;
   select count(*), sum(v) from fact join dim on fact.k = dim.k;
   +----------+--------------------+
   | count(*) | sum(fact.v)        |
   +----------+--------------------+
   | 40000    | 20121.308621485692 |
   +----------+--------------------+
   1 row(s) fetched.
   Elapsed 0.031 seconds.
   
   +----------+--------------------+
   | count(*) | sum(fact.v)        |
   +----------+--------------------+
   | 40000    | 20121.308621485692 |
   +----------+--------------------+
   1 row(s) fetched.
   Elapsed 0.017 seconds.
   
   +----------+-------------------+
   | count(*) | sum(fact.v)       |
   +----------+-------------------+
   | 40000    | 20121.30862148569 |
   +----------+-------------------+
   1 row(s) fetched.
   Elapsed 0.011 seconds.
   
   +----------+-------------------+
   | count(*) | sum(fact.v)       |
   +----------+-------------------+
   | 40000    | 20121.30862148569 |
   +----------+-------------------+
   1 row(s) fetched.
   Elapsed 0.011 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..1927333], [tmp/fact.parquet:1927333..3854666], 
[tmp/fact.parquet:3854666..5781999], [tmp/fact.parquet:5781999..7709332], 
[tmp/fact.parquet:7709332..9636665], ...]}, 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 AND 
k_null_count@1 != row_count@2 AND KEY_RANGE_BITMAP(k_min@3, k_max@0, 
20000/1048576 buckets), required_guarantees=[], metrics=[output_rows=200.0 K, 
elapsed_compute=9.65ms, output_bytes=25.0 MB, output_batches=200, 
files_ranges_pruned_statistics=12 total → 12 matched, 
row_groups_pruned_statistics=2.00 K total → 200 matched, 
row_groups_pruned_bloom_filter=200 total → 200 matched, 
page_index_pages_pruned=200 total → 200 matc
 hed, page_index_rows_pruned=200.0 K total → 200.0 K matched, 
limit_pruned_row_groups=0 total → 0 matched, batches_split=0, 
bytes_processed=22.1 MB, bytes_scanned=2.1 MB, 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=64.23µs, 
metadata_load_time=345.27µs, page_index_eval_time=1.33ms, 
row_pushdown_eval_time=36ns, statistics_eval_time=896.06µs, 
time_elapsed_opening=6.99ms, time_elapsed_processing=24.90ms, 
time_elapsed_scanning_total=224.99ms, time_elapsed_scanning_until_data=25.01ms, 
output_rows_skew=1.19%, scan_efficiency_ratio=9.7% (2.24 M/23.13 M)] |
   ...
   ```
   
   ## Are there any user-facing changes?
   
   <!--
   If there are user-facing changes then we may require documentation to be 
updated before approving the PR.
   
   If there are any breaking changes to public APIs, please add the `api 
change` label.
   -->
   


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: [email protected]

For queries about this service, please contact Infrastructure at:
[email protected]


---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]

Reply via email to