viirya opened a new pull request, #5778:
URL: https://github.com/apache/datafusion-comet/pull/5778
## Which issue does this PR close?
Closes #5777.
## Rationale for this change
`hash_list_array!` is the fallback for a list whose elements are not
primitives, which is the
path `array<struct<..>>` takes. It sliced a one-element Arrow array per
element and re-entered
the hash dispatch for it, so each element cost an allocation plus a full
type match, while a
list of primitives goes through a vectorized path and avoids both.
That is the shape behind
`spark.comet.shuffle.native.partitioning.hash.nested.enabled`
defaulting to off in #5567.
## What changes are included in this PR?
Spark chains the element hashes in order, so the elements of one row still
have to be hashed in
sequence, but the allocation and dispatch do not have to happen per element.
This hashes one
element per row at a time in a single batched call, seeding each slot with
the running hash of
its row — exactly what the per-element call did. Rows are independent, so
the dispatch count
drops from the total element count to the length of the longest list.
Two details make it a win rather than a trade:
- **Surviving rows are carried forward** instead of the batch being
rescanned each pass. A row
never becomes alive again once exhausted. Rescanning costs rows x
longest-list, which for one
long list among short ones is almost all wasted — 8192 rows where one
holds 1024 elements and
the rest hold one would scan 8.4M slots to hash 9215 elements.
- **A simpler walk when every row is non-null with the same length**,
recorded in the same
initialisation pass. No row drops out early there, so the survivor
bookkeeping would be pure
overhead; without this the uniform shapes regressed 6-8%.
The gather is indexed by the list's own offset width, since a `LargeList`
can hold more than
`u32::MAX` elements and narrowing the positions would silently wrap and hash
the wrong elements.
The implementation this replaces used `usize` throughout, so this keeps that
property.
## How are these changes tested?
Measured with the `hash` benchmark already on main from #5765, so the
numbers can be reproduced
against `main` rather than against a benchmark arriving with the change:
| case | main | this PR |
|---|---|---|
| `array<int32>` x10 | 146 µs | 147 µs |
| `array<struct<..>>` x10 | 10162 µs | **729 µs** |
| `array<struct<..>>` skewed | 1192 µs | **568 µs** |
8192 rows, Apple M4 Max. The primitive-element case is untouched, as
expected — it does not go
through this path.
The hash values must not move, so the tests are mostly about that. Golden
values for
`array<struct<..>>` pin empty lists, null lists, null struct elements, null
fields and repeated
values; a separate test asserts element order still changes the hash. Then
the cases this
implementation introduces: skewed lengths and descending lengths (so the
uniform-length path is
not taken), `LargeList` against `List` on the same data, `array<array<int>>`
where the element is
itself a list, rows exhausting on different passes, a sliced list whose
first offset is non-zero,
and a null list that still covers a non-empty element range. An off-by-one
in the survivor
condition fails several of them.
`cargo test -p datafusion-comet-spark-expr` passes 729 + 7,
`CometHashExpressionSuite` 40 and
`CometNativeShuffleSuite` 53. `cargo fmt --check` and `clippy -D warnings`
are clean.
## Additional context
This does not flip the config default. `array<struct<..>>` as a partitioning
key is now faster
than letting Spark do the shuffle rather than slower, but I would rather
change the default in a
separate PR with shuffle-level numbers than fold it in here.
Further headroom exists and is not taken here: profiling the remaining time
puts about 38% in
Arrow `take`, most of that copying string payloads. Removing it means
hashing fields directly at
gathered indices, which requires making the shared hash dispatch index-aware
— a much wider
change to a macro whose correctness decides partition assignment.
--
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]
