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]
