Repository navigation
Conversation
Codecov Report❌ Patch coverage is Additional details and impacted files@@ Coverage Diff @@
## main #26249 +/- ##
========================================
Coverage 82.92% 82.92%
========================================
Files 1154 1154
Lines 457790 458001 +211
Branches 457790 458001 +211
========================================
+ Hits 379637 379815 +178
- Misses 55135 55162 +27
- Partials 23018 23024 +6 ☔ View full report in Codecov by Harness. 🚀 New features to boost your workflow:
|
There was a problem hiding this comment.
From what I can gather, I like the approach of a specialized partitioning expression and after looking at the code for some time it makes sense. I also can't think of an easy way from the top of my head to improve the CASE evaluation in such a way that this specialized expression isn't necessary.
I've left a few notes that you can think of integrating. As I am not too familiar with the joining ecosystem I can't really approve it but to me the approach makes sense.
I think the only thing where this could be counter-productive is, if the data source can apply pruning with the CASE expression but not with the new special-purpose expression. I am not sure whether this happening or is even worth doing. Probably an optimization that transforms small multi-partition build sides into an IN LIST expression is the best way for this scenario.
| }; | ||
| merged_bounds = match (merged_bounds, real_partition_ids.len()) { | ||
| (None, 1) => Some(bounds.clone()), | ||
| (Some(merged), _) => merged.union(&bounds), |
There was a problem hiding this comment.
Is it expected that the union fails here or would that be a bug?
In case of the latter, maybe returning an error would be better than silently dropping it.
|
@tschwarzinger I'm actually thinking more about this and how I can get away from specialization that requires adding a whole new Expr. I was thinking about maybe folding this functionality in to https://docs.rs/datafusion/latest/datafusion/physical_plan/joins/struct.HashTableLookupExpr.html and just slightly modifying the existing operator 🤔 I'll get back to this PR later today or tomorrow, but, that's where my head is at right now. |
We already emit in list expressions for small builds sides, is there something I'm missing? |
I was able to reproduce this and generate a table + query (with the help of my buddy claude) to get pruning in the CASE expr working, but, not working in my specialization: Parquet file information: Results of pruning: Where the query is like so (including building parquet files): Going to continue playing with this and getting it to a much better state. |
What I meant was over multiple partitions. I am not sure whether we have anything there. For example: becomes |
|
Exploration also needs to consider number of partitions involved (i've looked at implementations before that are faster at 1->64 partitions but get slower after that), selectivity of the join, size of each side, pruning potential, size/length of the values being hashed, etc. There are many factors 😓 |
|
Thank you for investigating that! From what I can tell, the pruning for hash lookups is done via |
adriangb
left a comment
There was a problem hiding this comment.
Thank you for this PR, @devanbenz.
I ran some benchmarks, results look good. TPC-H SF100, CPU time vs main, lower is better. N is the number of partitions:
| Query | N=16 | N=128 |
|---|---|---|
| Q3 | 0.74 | 0.52 |
| Q5 | 0.72 | 0.46 |
| Q9 | 0.64 | 0.38 |
| Q21 | 0.67 | 0.41 |
This PR also looses pushdown:
-- 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;Probe DataSourceExec metrics:
main |
this PR | |
|---|---|---|
| predicate | CASE hash_repartition % 16 WHEN 0 THEN ... AND hash_lookup ... |
k@0 >= 0 AND k@0 <= 1800499 AND partitioned_hash_lookup(16) |
| pruning_predicate | min/max AND KEY_RANGE_BITMAP(...) per partition |
min/max only |
| row_groups_pruned_statistics | 100 total → 14 matched | 100 total → 91 matched |
| page_index_rows_pruned | 280.0 K total → 26.05 K matched | 1.82 M total → 1.80 M matched |
| output_rows | 26.05 K | 1.80 M |
Some more issues we should resolve before moving forward with this in review comments.
| lookups.push(PartitionLookup::Map(Arc::clone(map))); | ||
| Some(lookups) | ||
| } | ||
| _ => None, |
There was a problem hiding this comment.
This PR falls back to CASE at high target_partitions. That is where the CASE path costs the most: its cost grows about linearly with N (microbenchmark: about 6 / 50 / 150 ns/row at N=1 / 16 / 128).
Since this arm returns None as soon as one partition is not a hash map, and at high N each partition gets few keys, so some partitions pick InList (the per-partition 150-distinct threshold). Then the whole filter falls back to CASE hash_repartition % N.
MRE (uses probe.parquet from the other comment):
SET datafusion.execution.target_partitions = 128;
SET datafusion.optimizer.hash_join_single_partition_threshold = 0;
SET datafusion.optimizer.hash_join_single_partition_threshold_rows = 0;
EXPLAIN ANALYZE SELECT count(*)
FROM (SELECT value AS k FROM generate_series(1, 19200)) b
JOIN probe p ON b.k * 97 = p.k;| HASH_JOIN_SEED, | ||
| lookups, | ||
| )) as Arc<dyn PhysicalExpr>; | ||
| let bounds = merged_bounds |
There was a problem hiding this comment.
The merged bounds ignore canceled partitions; is that intended? It looks safe because a canceled partition’s rows have no consumer.
A CanceledUnknown partition pushes PartitionLookup::Const(true) but adds nothing to the bounds. The bounds predicate is then ANDed for every probe row. So a row routed to the canceled partition is removed if its key is outside the other partitions' range. That row can have a match in the canceled partition.
The CASE path keeps these rows, because each partition has its own bounds and a canceled partition gives true.
Possible fix (needs checking): do not use the merged bounds when a partition is canceled.
| let bounds = merged_bounds | |
| let bounds = merged_bounds.filter(|_| !has_canceled_unknown) |
| on_columns: Vec<PhysicalExprRef>, | ||
| routing_state: SeededRandomState, | ||
| lookup_state: SeededRandomState, | ||
| partitions: Vec<PartitionLookup>, |
There was a problem hiding this comment.
Related to the pruning regression (see the comment on shared_bounds.rs): this expression has no equivalent of HashTableLookupExpr::pruning_bitmap(), so PruningPredicate cannot use it.
Suggestion:
- Store one combined
Option<Arc<KeyRangeBitmap>>here. It is the union of the per-partition bitmaps. If any partition has no bitmap (or isConst(true)), useNone. - Expose it with
pruning_bitmap(), and letpruning_predicate.rshandlePartitionedLookupExprthe same way asHashTableLookupExpr.
pub(crate) struct PartitionedLookupExpr {
on_columns: Vec<PhysicalExprRef>,
routing_state: SeededRandomState,
lookup_state: SeededRandomState,
partitions: Vec<PartitionLookup>,
/// Union of the per-partition key-range bitmaps, for row group and page pruning.
pruning_bitmap: Option<Arc<KeyRangeBitmap>>,
}There was a problem hiding this comment.
@adriangb I'm actually thinking about utilizing HashTableLookupExpr instead of creating a new specialized operator--while keeping it's API surface the same/internals the same if possible so I can utilize PruningPredicate and all the other niceties of HashTableLookupExpr.
There was a problem hiding this comment.
Also see: #26249 (comment) where my operator failed to prune while CASE did it as expected.
There was a problem hiding this comment.
New changes allow pruning, I went ahead and utilized HashTableLookupExpr. Using
-- 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,
Building one hash table over every build partition's keys ran on a single thread at the dynamic filter barrier, so every probe stream waited on it (about 0.7s for a 15M key build at TPC-H SF10). HashTableLookupExpr now has a routed mode: each probe row is looked up in the table of the build partition its key was repartitioned to (hash or range routing), sharing the partitions' tables as they are. It stays a HashTableLookupExpr, so pruning keeps using its merged pruning bitmap. Removes union_hash_maps and JoinHashMapType::hash_values.
|
@adriangb When you get a moment could you please run your same benchmarks against this branch again. I pushed up some changes that I was cooking on over the past couple days to improve performance.
Locally I'm seeing:
* Caused by one outlier round (241.8s vs ~125s in the other two rounds). Using the following machine i7-12700K (20 threads, x86), 32 GB RAM With pushdown filters off - all tpch:
With pushdown filters on - all tpch
|
Work in progress/experiment, attempting to resolve a few performance issues with large CASE statements during dynamic filter scans. This commit reduces the amount of copies required during a predicate lookup. Specifically this was seen as a bottleneck on Q17 of TPC-H (~6s -> 2.20s): #26169 (comment)