andygrove opened a new pull request, #7057:
URL: https://github.com/apache/datafusion-comet/pull/7057
## Which issue does this PR close?
Part of #2545. This is a research draft for discussion, not a change to
merge as it stands.
## Rationale for this change
Comet's native shuffled hash join is DataFusion's `HashJoinExec`, whose
build side has to fit in the task's share
of the memory pool. When it does not, the task fails. That is why
`spark.comet.exec.forceShuffledHashJoin` is off by
default and has a build-size guard (#6384), even though a hash join is
faster than a sort-merge join when it fits.
DataFusion has an open design for spilling hash joins
(apache/datafusion#24768), but no agreed approach yet. The
sort-merge fallback (apache/datafusion#25217) is waiting on review, and a
refactor of the hash join stream is in
flight. Comet's case is much simpler than DataFusion's general one. Every
native shuffled hash join runs in
`Partitioned` mode, on one partition per task, and the output order contract
lives on the Spark side, not in
DataFusion's optimizer. This draft explores a Comet-only design for that
case, to find out what it costs and what
breaks. Parts of it could be offered upstream later.
## What changes are included in this PR?
The design is written up in
`docs/source/contributor-guide/spilling-hash-join-design.md`. In short:
- `SpillingHashJoinExec` (`native/operators/src/spilling_hash_join/`) wraps
the `HashJoinExec` that Comet already
plans. An equi-join splits exactly by a hash of its keys, for every join
type, so each bucket is joined by a stock
`HashJoinExec`, and the operator implements no join semantics of its own.
- While the build side fits, it runs that join unchanged.
- When the memory pool refuses, it routes both sides to 16 buckets,
destages the largest buckets to disk, joins
the buckets left in memory with the probe input, and then joins each
spilled bucket from its files. A bucket
that does not fit is split on the next bits of the hash. A bucket that
splitting cannot shrink, because most of
its rows share one key or have a null key, is joined in chunks, with a
replay of its probe rows for each chunk.
- Everything the operator and its inner joins hold is reserved under one
memory consumer. That includes what
`HashJoinExec` allocates without reserving it: its concatenated copy of
the build batches, and its chain of row
indices.
- Two configs. `spark.comet.exec.hashJoin.spill.enabled` is on by default,
so that CI runs the new join.
`spark.comet.exec.hashJoin.spill.maxBuildBufferBytes` forces spilling, for
tests. The serde enables spilling only
for shuffled hash joins that are not null-aware, do not use the join
runtime filter, and whose output order Spark
does not rely on.
- The Spark plan keeps `CometHashJoinExec`, and the native operator displays
as the `HashJoinExec` it runs, so plans
and golden files do not change. New metrics: `spilled_buckets`,
`split_buckets`, `chunked_buckets` and
`budget_overflow_bytes`, plus the spill metrics.
- `CometHashJoinMemoryBenchmark` compares Spark's sort-merge join with
Comet's sort-merge join, its hash join with
spilling disabled, and its hash join with spilling, across native memory
budgets.
- User docs for the join and its metrics.
It reviews best commit by commit.
## Findings so far
The benchmark joins 4M build rows with 20M probe rows on `local[4]`, with a
native pool per task from 1024 MiB
down to 64 MiB. The times are rough, because another benchmark was running
on the same machine (load 13 to 54).
Process CPU time is the steadier measure.
| Inner join, best ms / CPU s | Comet sort-merge join | Hash join, no
spilling | Hash join |
| --------------------------- | --------------------- |
---------------------- | -------------------- |
| 1024 MiB | 1061 / 4.4 | 803 / 3.4
| 861 / 3.2 |
| 192 MiB | 1391 / 6.4 | 818 / 3.3
| 1158 / 4.7, spilling |
| 128 MiB | 1422 / 6.2 | fails
| 1616 / 6.6 |
| 64 MiB | 1499 / 6.8 | fails
| 1692 / 6.9 |
- The spilling join completed for every join shape at every budget,
including where the join without spilling
fails. The shapes were inner, left, right and full outer, left anti, and
an inner join with a hot key.
- While the build side fits, it costs the same as the join without spilling.
- Reserving honestly moves the spill point. At 192 MiB the join without
spilling fits, but the spilling join
spills, mostly because it counts the concatenated copy of the build side
that `HashJoinExec` makes without
reserving it.
- When it spills heavily, its CPU time is about the same as the sort-merge
join's, so it is not faster yet.
Routing splits each 8192-row batch 16 ways, so spill files and inner joins
work on slivers of about 512 rows.
Coalescing them is the first thing to try.
- Two DataFusion facts came up along the way. A batch read back from an
Arrow IPC spill file holds more memory than
it did when it was written, as `HashJoinExec` counts it: an uncompressed
batch shares the buffer its message was
read into, metadata included, and an lz4 batch also keeps its compressed
message. And `HashJoinExec` counts its
final pass over the build rows as probe input in `input_rows`.
## Open questions
- Is the honest reservation of the concatenated copy worth the earlier spill
point, or should the build side be
concatenated as it arrives, so the copy is smaller?
- Should the join be on by default before it is faster than the sort-merge
join when it spills?
- How does this fit with #6785, which plans the forced shuffled hash join at
planning time, and with
`forceShuffledHashJoin.maxBuildSize` once joins can spill?
- Is 16 buckets per level the right fan-out?
## How are these changes tested?
- Rust tests compare the operator with a plain `HashJoinExec`, row for row,
for all 10 DataFusion join types, with
and without a residual filter. They cover each path: in memory, bucketed,
split again, chunked for a hot key and
for null keys, real pool refusals, every bucket spilled, and empty sides,
each with uncompressed and lz4 spill
files. They also check that no inner join reserves beyond its estimate,
that the operator's row counts are right,
that a failed acquire fails the join instead of spilling, and that
dropping the join deletes its spill files.
Four deliberate breakages of the operator were each caught.
- A planner test checks that the join is wrapped, that it keeps the
projection `swap_inputs` adds, and that a join
with the runtime filter is left alone.
- `CometSpillingHashJoinSuite` checks that under a native pool too small for
the build side the join fails without
spilling, and completes and matches Spark with it. It also checks that
every Spark join type matches Spark while
spilling, and which joins are eligible.
- `CometJoinSpillingSuite` runs all of `CometJoinSuite` with every native
shuffled hash join spilling all of its
build side.
- `CometJoinSuite`, `CometExecSuite` and `CometPlanEqualitySuite` pass on
Spark 4.1.
- Not yet run: the Spark SQL and Iceberg suites, TPC-H and TPC-DS, and
timings on a quiet machine.
--
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]
