wangyum opened a new pull request, #60048: URL: https://github.com/apache/spark/pull/60048
### What changes were proposed in this pull request? This PR shrinks the sparse index array of `LongToUnsafeRowMap` once at `optimize()` time when the dense conversion does not apply: the occupied `(key, address)` pairs are rehashed into the smallest power-of-two array that keeps the load factor invariant of `updateIndex` (`numKeys * 4 <= array.size()`). `write()`/`read()` are unchanged, so the wire format stays as-is and every executor now deserializes a right-sized index. A build that is already right-sized (unique keys, or any map that grew during the build) is a no-op, and the shrink never grows: at the `1 << 30` array-size ceiling a map may hold more keys than a minimal table could take, so it keeps its array. ### Why are the changes needed? The initial capacity comes from the estimated row count (SPARK-22170), but only distinct keys occupy slots, and the sparse array keeps a load factor `<= 0.5` headroom for probing. With many rows sharing few distinct keys and a key range too wide for the dense conversion, the finished map keeps a mostly-empty array that `write()` ships verbatim, so the headroom is paid for in driver memory, in the serialized broadcast payload, and again in every executor's deserialized copy. For example, 100M rows sharing 10K wide-range keys keep a ~2 GiB index that only needs 512 KiB (~4096x waste). The rehash cost at `optimize()` time is one sequential scan of the mostly-empty array plus `O(numKeys)` reinserts, paid once on the driver; for builds that already grew, the growth ladder lands on the minimal size and the shrink is a no-op. `ShuffledHashJoinExec` does not pass a size estimate and its growth ladder already lands on the minimal size, so this is effectively broadcast-only. ### Does this PR introduce _any_ user-facing change? No. Broadcast "data size" metrics now report smaller (more accurate) sizes for builds with duplicated keys, which can also let such broadcasts pass the hard broadcast size cap they previously exceeded after the build. No API, config, or wire format change. ### How was this patch tested? New tests in `HashedRelationSuite`, run in both on-heap and off-heap modes: - `LongToUnsafeRowMap shrinks an over-allocated sparse array at optimize time`: 99999 rows sharing 3 wide-range keys, asserting the exact post-shrink footprint, per-key value counts, `keys()`, and a no-op second `optimize()`. - `LongToUnsafeRowMap keeps a right-sized sparse array unchanged`: unique keys, no footprint change. - `LongToUnsafeRowMap serialization carries the shrunk sparse array`: bounds the serialized size and round-trips lookups. - `LongToUnsafeRowMap collapses an empty map to the minimal array`. `HashedRelationSuite` (56 tests), `BroadcastExchangeSuite` and `JoinSuite` (969 tests) pass locally. ### Was this patch authored or co-authored using generative AI tooling? Generated-by: GLM 5.3 -- 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]
