comphead commented on PR #6095:
URL:
https://github.com/apache/datafusion-comet/pull/6095#issuecomment-5769660944
Ran the partitioning benchmarks on this branch so the two approaches can be
compared on numbers,
since the PR description invites that. Short version: `RowGroups` is both
safer and **faster** than
#5449's `WholeBatch`, and the reason is the run-based index representation
rather than the placement
key.
## Method
`native/shuffle/benches/shuffle_writer.rs`, 8 batches of 8192 rows into 50
output partitions,
`CompressionCodec::None`. Two schema shapes with row and batch counts held
equal:
- **plain**: the flat 4-column schema that file already uses (int32, utf8,
date32, decimal128), ~34 B/row
- **nested**: 40 struct columns of depth 2, leaf `struct<a: int64, b: utf8,
c: float64>`, ~840 B/row
This is the `partitioning_only` microbench, so `insert_batch` only:
placement plus index buffering,
no flush, no IPC encode, no disk. `AUTO_GROUP_ROWS` resolves to 163 here
(8192 / 50).
## Results
| schema | `HashAll` (all cols) | `HashAll{1}` | `RowGroups` AUTO=163 |
`RowGroups` 8192 |
| ------ | -------------------- | ------------ | -------------------- |
---------------- |
| plain | 800.8 µs | 233.4 µs | **13.8 µs** | 9.3
µs |
| nested | 12.72 ms | 685.5 µs | **204.6 µs** |
202.6 µs |
Speedup over `HashAll`, measured in the same binary:
| schema | `HashAll{1}` | `RowGroups` AUTO | `RowGroups` 8192 | #5449
`WholeBatch` |
| ------ | ------------ | ---------------- | ---------------- |
------------------ |
| plain | 3.4x | **57.9x** | 85.8x | 11.3x -
22.2x |
| nested | 18.6x | **62.2x** | 62.8x | 47.2x -
48.8x |
The #5449 column is from a separate binary, so I normalised everything to
the `HashAll` baseline
measured alongside it. The two baselines agree closely, which is what makes
the comparison usable:
plain 800.8 µs here vs 782-785 µs there (+2.4%), nested 12.72 ms vs
12.77-12.84 ms (-0.4%).
Intervals were tight throughout, well under 1% on every row except the
smallest.
## Why this is faster than #5449, not just safer
`WholeBatch` assigns a whole batch to a partition but still appends one
`(batch, row)` pair per row
to `partition_indices`, so it keeps an O(rows) write and 8 bytes per row
against the spill
reservation. `PartitionIndices`' run shape here records `(batch, start,
len)` instead, which at
AUTO=163 is about 50 runs per batch rather than 8192 pairs. That is where
the extra ~30% on nested
and the 2.6x to 5x on plain come from.
So the run representation looks worth having independently of which
placement key wins.
## On `group_rows`
AUTO=163 versus a batch-sized 8192 is a wash on the nested schema (204.6 vs
202.6 µs, inside the
interval) and only about 1.5x on plain, where absolute times are small
enough that fixed costs
dominate. The AUTO default looks cheap on the shape that motivated this
work, so bounding imbalance
at 163 rows does not appear to cost much.
## One data point on the cheap-hash middle ground
`HashAll { max_hash_columns: 1 }` is 18.6x on nested, so hashing one column
instead of recursing
through forty captures a good part of the gap while staying content-derived
and needing none of the
determinism machinery. It is not a substitute though, for the reason your
`duplicate rows spread evenly` test pins: a low-cardinality leading column
collapses the
distribution, and positional placement bounds imbalance by `group_rows`
regardless of the data.
Worth knowing the number, not worth reaching for.
## Caveats, and one thing that needs fixing first
**Every end-to-end bench in that file panics during criterion warmup**,
including the pre-existing
hash and range ones on `main`:
```
shuffle write error: partition offsets were already published
```
One `ShuffleWriterExec` is built outside `b.iter()` and re-executed per
iteration, but
`PartitionOffsets` is a `OnceLock` that errors on a second `set`. CI does
not catch it because
`pr_benchmark_check.yml` only runs `cargo check --benches`. I fixed it
locally with `b.iter_batched`
to get the numbers above. Probably worth a separate PR, since it affects all
partitionings.
**I have no trustworthy end-to-end numbers.** The nested end-to-end bench
writes ~55 MB uncompressed
per iteration, so ~5.5 GB per bench function, and the machine I ran on was
at 99% disk. Across three
runs the `HashAll` nested figure drifted 55.8 to 44.0 to 38.6 ms with 15-19%
of samples flagged as
outliers, and the ratio moved between 1.03x and 1.72x. That path is
I/O-bound here and measures the
disk, not the change. Worth redoing on a quiet host, since
`partitioning_only` excludes the flush
and therefore does not exercise `RunIterator` at all, which is likely where
more of the win is.
**The nested fixture is constant-valued** (`1i64`, `"x"`, `1.0` on every
row), so it measures hash
cost and not hash distribution. `HashAll{1}` would send everything to one
partition on that data.
**The microbench needed local-only visibility changes** to reach
`MultiPartitionShuffleRepartitioner`
and friends, which this PR deliberately does not export. Those are not part
of any proposal here,
just measurement scaffolding.
--
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]
