devanbenz commented on code in PR #26249:
URL: https://github.com/apache/datafusion/pull/26249#discussion_r4320473111
##########
datafusion/physical-plan/src/joins/hash_join/partitioned_hash_eval.rs:
##########
@@ -458,6 +456,209 @@ impl PhysicalExpr for HashTableLookupExpr {
}
}
+/// Build-side lookup for one partition of a `Partitioned` hash join.
+pub(crate) enum PartitionLookup {
+ /// The partition's hash table; probe rows routed here are looked up in it.
+ Map(Arc<Map>),
+ /// The partition is known empty (`false`) or its contents are unknown
(`true`).
+ Const(bool),
+}
+
+/// Routes each probe row to the hash table of the build partition that would
+/// contain it and looks the row up there.
+///
+/// This replaces `CASE hash(keys) % n WHEN i THEN lookup_i ... END` for
+/// partitioned joins. `CaseExpr` evaluates each branch by filtering the batch
+/// down to that branch's rows and merging the partial results, which copies
+/// every column of the batch once per branch. This expression computes the
+/// routing and lookup hashes once, buckets row indices by partition and only
+/// gathers the 8-byte hashes, so the batch columns are never copied.
+pub(crate) struct PartitionedLookupExpr {
+ on_columns: Vec<PhysicalExprRef>,
+ routing_state: SeededRandomState,
+ lookup_state: SeededRandomState,
+ partitions: Vec<PartitionLookup>,
Review Comment:
New changes allow pruning, I went ahead and utilized `HashTableLookupExpr`.
Using
```sql
-- Data: probe sorted by k, small row groups and pages; build has 5,000
sparse keys.
COPY (SELECT value AS k, value * 2 AS v FROM generate_series(0, 1999999)
ORDER BY k)
TO '/tmp/mre/probe.parquet' OPTIONS (max_row_group_size 20000,
data_page_row_count_limit 2000);
COPY (SELECT value AS k FROM generate_series(0, 1999999) WHERE value %
200000 < 500)
TO '/tmp/mre/build.parquet';
CREATE EXTERNAL TABLE probe STORED AS PARQUET LOCATION
'/tmp/mre/probe.parquet';
CREATE EXTERNAL TABLE build STORED AS PARQUET LOCATION
'/tmp/mre/build.parquet';
SET datafusion.execution.target_partitions = 16;
SET datafusion.execution.parquet.pushdown_filters = false;
SET datafusion.optimizer.hash_join_single_partition_threshold = 0;
SET datafusion.optimizer.hash_join_single_partition_threshold_rows = 0;
EXPLAIN ANALYZE SELECT count(*) FROM build b JOIN probe p ON b.k = p.k;
```
Here is the `DataSourceExec`
```
| | DataSourceExec: file_groups={16 groups:
[[tmp/mre/probe.parquet:0..657492], [tmp/mre/probe.parquet:657492..1314984],
[tmp/mre/probe.parquet:1314984..1972476],
[tmp/mre/probe.parquet:1972476..2629968],
[tmp/mre/probe.parquet:2629968..3287460], ...]}, projection=[k],
output_ordering=[k@0 ASC NULLS LAST], file_type=parquet,
predicate=DynamicFilter [ k@0 >= 0 AND k@0 <= 1800499 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 <= 1800499 AND
k_null_count@1 != row_count@2 AND KEY_RANGE_BITMAP(k_min@3, k_max@0,
2508/1048576 buckets), required_guarantees=[], metrics=[output_rows=26.05 K,
elapsed_compute=18.74ms, output_bytes=896.0 KB, output_batches=14,
files_ranges_pruned_statistics=16 total → 16 matched,
row_groups_pruned_statistics=100 total → 14 matched,
page_index_pages_pruned=140 total → 14 matched, page_index_rows_pruned=280.0 K
total
→ 26.05 K matched, limit_pruned_row_groups=0 total → 0 matched,
batches_split=0, bytes_processed=10.0 MB, bytes_scanned=328.0 KB,
file_open_errors=0, file_scan_errors=0, files_opened=16, files_processed=16,
num_predicate_creation_errors=0, page_index_load_skipped=5,
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=82.82µs,
metadata_load_time=2.66ms, page_index_eval_time=2.14ms,
row_pushdown_eval_time=43ns, statistics_eval_time=5.01ms,
time_elapsed_opening=54.47ms, time_elapsed_processing=78.27ms,
time_elapsed_scanning_total=42.34ms, time_elapsed_scanning_until_data=32.50ms,
output_rows_skew=37.62%, scan_efficiency_ratio=3.19% (335.9 K/10.52 M)] |
| |
```
More specifically the pruning metrics:
```
metrics=[output_rows=26.05 K, elapsed_compute=18.74ms, output_bytes=896.0
KB, output_batches=14, files_ranges_pruned_statistics=16 total → 16 matched,
row_groups_pruned_statistics=100 total → 14 matched,
page_index_pages_pruned=140 total → 14 matched, page_index_rows_pruned=280.0 K
total → 26.05 K matched,
```
--
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]
