andygrove commented on issue #17494:
URL: https://github.com/apache/datafusion/issues/17494#issuecomment-6234744182
Here is a reproduction of what makes q72 slow, in DataFusion alone, from the
TPC-DS runs for
apache/datafusion-comet#7057. In Comet at scale factor 100, with 16
concurrent Spark tasks on DataFusion 55.2, q72
takes 378 s when its join of `catalog_sales` and `inventory` is a sort-merge
join, and did not finish in 40 minutes
as a `HashJoinExec`. The tasks of that stage take 179 s on average with the
sort-merge join, and over 1,000 s with
the hash join.
The join is on the item alone, so each probe row has many build rows with
its key, and the residual filter
`inv_quantity_on_hand < cs_quantity` keeps a few percent of them. This
script builds tables of that shape: 64M
build rows over 640K keys, about 100 rows per key, and 96M probe rows, with
a filter that keeps 6% of the 9.6B
candidate pairs. With 16 partitions, each partition's build side has about
4M rows, like a Comet task's.
```sql
-- A join shaped like TPC-DS q72 catalog_sales x inventory: each key has
about 100 rows on the build side and
-- a residual filter on both sides keeps about 5% of the candidate pairs. 16
partitions, each with a build
-- side of about 4M rows, like the 16 concurrent tasks of the Comet run.
SET datafusion.execution.target_partitions = 16;
-- inventory: 64M rows over 640K items, quantity on hand 0..999, and two
payload columns
CREATE TABLE inv AS
SELECT value % 640000 AS item, (value * 7919) % 1000 AS qty, value % 15 AS
warehouse, value % 261 AS week
FROM generate_series(1, 64000000);
-- catalog_sales: 96M rows over the same items, quantity 1..100, and two
payload columns
CREATE TABLE cs AS
SELECT (value * 104729) % 640000 AS item, (value * 31) % 100 + 1 AS qty,
value % 1920800 AS cdemo,
value % 7200 AS hdemo
FROM generate_series(1, 96000000);
SELECT count(*) FROM inv;
SELECT count(*) FROM cs;
SET datafusion.optimizer.prefer_hash_join = true;
EXPLAIN SELECT count(*), sum(inv.warehouse), sum(inv.week), sum(cs.cdemo),
sum(cs.hdemo)
FROM cs JOIN inv ON cs.item = inv.item AND inv.qty < cs.qty;
SELECT count(*), sum(inv.warehouse), sum(inv.week), sum(cs.cdemo),
sum(cs.hdemo)
FROM cs JOIN inv ON cs.item = inv.item AND inv.qty < cs.qty;
SELECT count(*), sum(inv.warehouse), sum(inv.week), sum(cs.cdemo),
sum(cs.hdemo)
FROM cs JOIN inv ON cs.item = inv.item AND inv.qty < cs.qty;
SET datafusion.optimizer.prefer_hash_join = false;
EXPLAIN SELECT count(*), sum(inv.warehouse), sum(inv.week), sum(cs.cdemo),
sum(cs.hdemo)
FROM cs JOIN inv ON cs.item = inv.item AND inv.qty < cs.qty;
SELECT count(*), sum(inv.warehouse), sum(inv.week), sum(cs.cdemo),
sum(cs.hdemo)
FROM cs JOIN inv ON cs.item = inv.item AND inv.qty < cs.qty;
SELECT count(*), sum(inv.warehouse), sum(inv.week), sum(cs.cdemo),
sum(cs.hdemo)
FROM cs JOIN inv ON cs.item = inv.item AND inv.qty < cs.qty;
```
`datafusion-cli -f` with the `release-nonlto` profile on an AMD 7950X3D (16
cores), two runs each, on an otherwise
idle machine:
| Join | 55.2.0 | main (484183dc3) |
| --------------- | -------------- | ---------------- |
| Hash join | 61.1 s, 60.2 s | 62.0 s, 62.6 s |
| Sort-merge join | 11.8 s, 11.4 s | 11.8 s, 12.6 s |
With sparse keys (`item * 7919 + 1000000000000` on both sides), so that the
build side is a `JoinHashMap` instead of
an `ArrayMap`, main takes 76.9 s and 77.1 s with the hash join and 16.9 s
and 16.1 s with the sort-merge join.
`perf` on main: 85% of the samples are in
`ArrayMap::get_matched_indices_with_limit_offset`, with `traverse_chain`
inlined into it, and 11% in collecting the index `Vec`s. With sparse keys,
68% are in
`JoinHashMapU32::get_matched_indices_with_limit_offset` and 20% in
`equal_rows_single_col`.
The chain walk is bound by memory latency once the build side no longer fits
in the cache. With one partition and
a build side of 4M rows (4M build rows and 6M probe rows over 40K keys,
`target_partitions = 1`), the hash join
takes 2.7 s on a core with 96 MB of L3 and 7.8 s on one with 32 MB. With 16
partitions sharing the caches, as on a
busy executor, each partition gets a small part of them. That fits the
suggestions above: flattening the chains into offsets and
lengths when the build side has repeated keys, or a radix-partitioned join
(#18939).
--
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]
