bench: measure nested types as native shuffle hash partitioning keys - #5788
Conversation
Nothing covered the shapes admitted by `spark.comet.shuffle.native.partitioning.hash.nested.enabled`. `shuffleArrayBenchmark` and `shuffleStructBenchmark` do repartition on a nested column but have no native-shuffle case, because the gate rejected nested keys when they were written, and `shuffleDeeplyNestedBenchmark` calls `repartition(n)` with no key, which is round robin rather than hash partitioning. Adds `shuffleNestedHashKeyBenchmark` with a native-shuffle case that enables the config, so it measures the native hashing path rather than a silent fallback. The shapes separate the two code paths in the native Murmur3 kernel: a list whose elements are primitives is vectorized, while a list whose elements are nested falls through to `hash_list_array!`, which slices a one-element array and re-enters `create_murmur3_hashes` for every element. Measured on an Apple M4 Max, local[5], 1M rows (best time, relative to Spark doing the whole shuffle): struct<int, string> 5 partitions 79ms -> 41ms 1.9x struct<int, string> 201 partitions 126ms -> 89ms 1.4x array<int> 5 partitions 76ms -> 52ms 1.5x array<int> 201 partitions 145ms -> 101ms 1.4x struct<array<int>, string> 5 partitions 100ms -> 68ms 1.5x array<struct<int, string>> 5 partitions 164ms -> 561ms 0.3x array<struct<int, string>> 201 partitions 250ms -> 647ms 0.4x struct<map<string, int>, int> 5 partitions 121ms -> 97ms 1.2x struct<map<string, int>, int> 201 partitions 160ms -> 155ms 1.0x The native case is not silently falling back to Spark's shuffle. Rerunning the `array<struct<..>>` case at 5 partitions separates the three Comet cases, at 761ms (Spark shuffle), 1059ms (JVM shuffle) and 2340ms (native shuffle), so the native path is distinct and is the slow one. Absolute times move with machine load, but the ordering and the set of shapes that win or lose reproduce. So `array<struct<..>>` is the one shape that loses, at 0.3-0.4x of leaving the shuffle to Spark, which is what the config defaults to off for. Every other shape measured is faster natively. That is the fallback path tracked in apache#5777, so these numbers are the before side of that comparison rather than a standing property of the kernel. A map nested in a struct is not affected, because the kernel specializes common scalar key/value pairs. Co-authored-by: Claude Code <noreply@anthropic.com>
sunchao
left a comment
There was a problem hiding this comment.
Correctness
Reviewed 49a7196910ab210095a9393dc46c0e325d8e5cc7 against 17f54da8ca5cb0ad5dbe8357b6e037fef8a0db2c. This is a benchmark-only addition: one helper and five key shapes at 5 and 201 requested partitions, each compared across Spark, Comet with Spark shuffle, Comet JVM shuffle and Comet native shuffle. Existing array/struct benchmarks lack a native arm, while the deeply nested benchmark uses keyless round-robin repartitioning. The new explicit repartition(..., Column("k")) reaches hash partitioning and gives the nested-key gate a useful regression baseline.
One P2 needs addressing: the map-containing shape is unconditionally registered even on Spark 3.5, where native map hash partitioning is deliberately rejected. It therefore produces a Spark-shuffle timing under the native label. The inline comment asks to skip unsupported native cases and verify native exchange selection outside the timer. The existing map-shuffle test already expects zero native exchanges below Spark 4.0. There are no existing review discussions to duplicate.
The maintained Spark 3.5/4.0 sources confirm ordered field/element hash chaining, null-as-seed behavior and positive-modulus partition assignment. Comet uses the same seed, 42. Spark 4.0 recursively inserts MapSort for a map-containing repartition key, and this benchmark's scalar string map keys fit Comet's supported normalization path. Collated strings, interval leaves and non-scalar map keys remain outside the selected shapes and have separate fallback gates. Maintained Spark 3.4/4.1 sources are unavailable, so this source comparison does not qualify those versions.
The keys are built from a persisted nonconstant INT column. Reducing the mixed Long values with pmod before the cast avoids ANSI overflow. A separate arithmetic transcription of the input generator yields 1,048,576 rows, 649,586 distinct keys and at most eight rows per key. This is arithmetic evidence, not a Spark/native run or a reducer-distribution measurement. All generated values are non-null, arrays contain ten repeated elements and maps contain one entry. The benchmark therefore covers these fixed shapes, not null/empty/variable-length or multi-entry-map behavior. Existing native shuffle tests cover partition assignment, null levels, map entry order and unsupported types. Those tests were inspected, not executed for this PR.
The benchmark compile/lint job passed on merge 00f7c31fafa34ced8515432ec5123a3334e8a4aa. Its parents are the assigned BASE/HEAD, and its tracked tree equals HEAD. The log confirms Spark 4.0 compilation of the test sources plus Scala lint and Rust format/clippy/benchmark checks. At 2026-09-09T06:36:48.648551+00:00, eight checks succeeded and eleven were skipped, with none failed or running. CI did not execute the benchmark or the native/JVM correctness suites. I did not run local product builds or tests.
Performance
The shapes distinguish the intended code paths: primitive arrays use the typed element loop, arrays of structs use the recursive per-element slicing path, and the string-to-int map uses its specialized branch when native admission succeeds. Keeping the harness separate from the tracked kernel issue gives later changes a stable comparison point.
Input-file generation happens before the timer. The no-op datasource consumes the output, so the action includes SQL planning, Parquet reads, key construction, shuffle writing/reading and sink consumption. This is a full-operation comparison rather than an isolated hash-kernel measurement. All four arms share input and key expressions within each shape. The existing timer uses at least five iterations with 500 ms warmup and measurement floors, runs arms sequentially, and reports best time alongside average and deviation.
The PR's M4 Max results remain author-reported: 1.4–1.9x for the simple struct/array examples and 0.3–0.4x for arrays of structs. The PR explicitly omits one unrecorded cell and reports load-sensitive rerun times. There is no matched raw output, build identity or physical-plan receipt here to independently verify those benefits. Timing differences alone do not establish native execution. Also, a ratio against the all-Spark arm includes scan/projection differences and does not isolate changing the nested-key setting in an otherwise fixed Comet plan. The native-plan check requested above is necessary for this harness to detect the intended regressions reliably.
Design
The shared helper keeps data preparation, queries and partition counts aligned across four explicit execution modes. A small shape table makes the intended coverage easy to extend. The benchmark changes no production gate or hashing behavior. The needed adjustment is at case admission: reject or skip version-incompatible native cases, then validate the selected exchange before measuring. That preserves the conservative map normalization rule while making the reported label trustworthy.
Abstraction & complexity
The helper and shape list fit the existing benchmark framework and introduce no new execution or reporting layer. A small admission check can address the finding without adding a separate harness. I found no other actionable abstraction or complexity issue in the 94-line addition.
| "array<struct<int, string>>" -> | ||
| "ARRAY_REPEAT(named_struct('a', c1, 'b', CAST(c1 AS STRING)), 10)", | ||
| "struct<map<string, int>, int>" -> | ||
| "named_struct('m', MAP(CAST(c1 AS STRING), c1), 'i', c1)") |
There was a problem hiding this comment.
Correctness
[P2] Skip unsupported map-key native benchmark cases
Could we restrict this map-containing native case to supported Spark versions and check that the planned exchange is CometNativeShuffle before timing it? On the supported Spark 3.5 profile, nativeShuffleFailureReasons recursively reaches this MapType and rejects it because isSpark40Plus is false, even with the nested config enabled. In native-only mode the rule retains Spark's ShuffleExchangeExec, but this helper still records the run as Comet (Native Shuffle). The existing map-key shuffle test explicitly expects zero native exchanges below 4.0. These timings cannot detect native map-hashing regressions. A version guard or explicit skip plus an untimed plan check would keep the label accurate.
There was a problem hiding this comment.
Addressed in 8d9eede. The benchmark now skips native cases with map-containing keys on Spark versions below 4.0. For supported cases, it checks for exactly one CometNativeShuffle exchange outside the timer and records the physical plan.
Verified on Spark 3.5.9 at both 5 and 201 partitions: struct keys use native shuffle, while map-containing keys explicitly skip the native arm. The full Spark 4.1.3 run passed all 18 native exchange checks.
andygrove
left a comment
There was a problem hiding this comment.
The doc comment on shuffleNestedHashKeyBenchmark says the Spark arm is "what the config's default trades against", and I do not think that holds under the default spark.comet.shuffle.mode=auto. With spark.comet.shuffle.native.partitioning.hash.nested.enabled off, nativeShuffleFailureReasons in CometShuffleExchangeExec.scala records a reason for the nested key, and then columnarShuffleFailureReasons comes back empty for all five of these shapes, because nested types pass its serializable-type check and the partition expression is a plain attribute. So the shuffle lands on CometColumnarShuffle, not on Spark's ShuffleExchangeExec. Only an explicit mode=native falls all the way back to Spark. That makes Comet (JVM Shuffle) the arm the default should be judged against. Could the summary table carry that column as well? You already have it for array<struct<..>> from the rerun at 1059ms against 2340ms, but for the shapes reported as 1.4x to 1.9x wins over Spark it is not clear whether native beats what Comet already does today with the config off.
The map shape is MAP(CAST(c1 AS STRING), c1), so every row holds exactly one entry. On Spark 4.x the repartition key becomes named_struct('m', mapsort(k.m), 'i', k.i) via InsertMapSortInRepartitionExpressions, and a mapsort over a single entry does nothing, while the specialized (Utf8, Int32) branch in hash_funcs/utils.rs runs its inner loop once per row. So the conclusion that a map nested in a struct is unaffected rests on a shape that does close to no map work, and the normalization Spark inserts is the part that stops being free at realistic entry counts. Would a map with several entries per row be a better shape to measure here?
CI is green on 49a7196, including the benchmark compile and lint job.
sunchao
left a comment
There was a problem hiding this comment.
Correctness
Rechecked unchanged head 49a7196910ab210095a9393dc46c0e325d8e5cc7 against base 17f54da8ca5cb0ad5dbe8357b6e037fef8a0db2c after the new review. [P2] The existing finding on unsupported map-key native cases remains: Spark 3.5 rejects the map-containing native arm even with nested hashing enabled, and the helper still neither skips it nor checks the selected exchange. I have no additional inline finding to add.
I traced the default alternative through shuffleSupported and both exchange eligibility checks. With the benchmark's CometShuffleManager, Comet execution/shuffle enabled, and otherwise default support settings, disabling spark.comet.shuffle.native.partitioning.hash.nested.enabled gives the following source-level selection at both requested partition counts:
| Key shape | auto, nested off |
native, nested off |
native, nested on |
|---|---|---|---|
struct<int,string> |
Comet JVM shuffle | Spark exchange | Comet native shuffle |
array<int> |
Comet JVM shuffle | Spark exchange | Comet native shuffle |
struct<array<int>,string> |
Comet JVM shuffle | Spark exchange | Comet native shuffle |
array<struct<int,string>> |
Comet JVM shuffle | Spark exchange | Comet native shuffle |
struct<map<string,int>,int> |
Comet JVM shuffle | Spark exchange | Native on Spark 4.0, Spark exchange on 3.5 |
All five schemas pass the columnar serializer gate. The first four partition expressions remain attributes. For the map shape, maintained Spark 4.0 inserts named_struct('m', mapsort(k.m), 'i', k.i), whose scalar string map key is supported by CometMapSort. Thus the fifth shape also passes the JVM exchange check. Explicit jvm mode selects that exchange regardless of the nested flag. With auto and the flag on, the first four can select native on 3.5/4.0, while the map shape remains JVM on 3.5. These are code-path conclusions, not newly captured physical plans. Maintained Spark 3.4/4.1 sources remain unavailable.
At the 2026-09-09T12:43:26.878Z cutoff, eight checks succeeded and eleven were skipped. I re-read the benchmark compile/lint log: it checks out merge 00f7c31, whose parents are this base/head and whose complete tracked tree equals the head. It compiles the Spark 4.0 test sources and runs lint checks with tests skipped. It does not establish benchmark execution, native exchange selection or performance. No local product build, runtime test or benchmark was run in this follow-up.
Performance
The new review's baseline request is supported. Could the helper's comment and summary table use the already-present Comet (JVM Shuffle) arm as the default auto/nested-off comparison, with matched JVM/native results for every shape at both partition counts? The all-Spark arm disables Comet scan and projection as well. Even under explicit native mode with nested hashing off, the Spark exchange can retain Comet upstream execution, corresponding to the Comet (Spark Shuffle) arm. A win over all-Spark therefore does not establish a win from enabling nested hashing in either configuration.
The singleton-map limitation also warrants the requested additional evidence. MAP(CAST(c1 AS STRING), c1) gives one entry per row. The native (Utf8, Int32) specialization avoids recursive per-entry array slices, but still hashes each key/value pair in a loop. It does not eliminate work proportional to entry count. Also, spark_map_sort has no singleton-map fast path: it builds indices and offsets, calls sorting, and takes the entries. One entry requires no meaningful ordering comparison, but normalization is not free.
Could we retain the singleton control and add maps with several distinct entries, varied cardinalities and differing input key orders, then report matched JVM/native timings before generalizing that maps are unaffected? That would exercise normalization and the specialized hash loop at representative sizes. The reported rerun of 2,340 ms native versus 1,059 ms JVM for arrays of structs is about 2.2x slower within that reported cohort. It remains author-provided evidence and should not be mixed with the earlier, differently loaded timing table.
Design
The existing four arms already provide both relevant Comet alternatives. Clarifying which one answers the configuration question is sufficient without adding another harness. The remaining admission fix should pair version-aware skips with an exchange check outside the timer, so labels and reported comparisons describe the path that actually ran. The external review already raises the baseline and map-coverage requests, so this follow-up builds on that discussion rather than duplicating it inline.
Abstraction & complexity
There is no new abstraction change since the prior review. The shared helper and shape list can accommodate additional map cardinalities while keeping inputs and partition counts aligned across arms. Keep the baseline interpretation, unsupported-case handling and plan verification explicit in that helper. No additional framework or production change is needed for these benchmark corrections.
Skip native map-key cases before Spark 4.0 and verify the native exchange before timing. Compare against the JVM shuffle baseline, add variable-size maps in both input orders, and provide a nested-hash-only entry point. Validated Spark 4.1.3 with 18 groups / 72 benchmark cases and native plan checks. Spark 3.5.9 clean build and 14-case smoke run confirm native struct admission and explicit native map skips at 5 and 201 partitions.
|
@andygrove Thanks, both points are addressed in 8d9eede and the updated PR description. The comment now identifies Comet JVM shuffle as the baseline for default I retained the singleton-map control and added maps with 2–10 and 2–50 distinct entries per row, each in forward and reversed input order. The results do not support a general claim that maps are unaffected: for the 2–50-entry reversed case at 201 partitions, native took 2,449 ms versus 2,107 ms for JVM shuffle. I removed that claim and documented that these end-to-end, sequential measurements do not isolate sorting or hashing costs. |
sunchao
left a comment
There was a problem hiding this comment.
Correctness
Rechecked 8d9eede019d09e18c85c18f8b383d3e2c2cd1ede against 17f54da8ca5cb0ad5dbe8357b6e037fef8a0db2c, including the full one-file contribution and the increment from review 5154588398. The P2 is addressed. Recursive map detection omits the native arm below Spark 4.0. Admitted cases require exactly one CometNativeShuffle exchange before timing, and the check and timed action use the same query, partition count and native configuration. The inherited plan collector traverses adaptive plans and query stages.
This matches maintained Spark 3.5/4.0 behavior: hash repartitioning uses the supplied key, while Spark 4.0 recursively normalizes map-containing keys with MapSort and Comet declines that native path on 3.5. The new maps have distinct string keys and aligned integer values. Reversing both input arrays preserves the pairs, and 2 + pmod(c1, maxEntries - 1) supplies the stated 2–10 and 2–50 ranges. The bounded c1 keeps the added integer arithmetic within range under ANSI mode. I found no new or remaining P1/P2.
At 2026-09-09T19:53:06.976426+00:00, eight checks passed and eleven were skipped. The benchmark compile/lint job compiled the Spark 4.0 test sources with tests skipped. It ran merge fdd5d4b1, whose parents are newer base 424c31aa and this head, rather than the assigned base. The benchmark file is identical in that merge, but the complete trees differ. This is compilation evidence for the changed source, not benchmark execution or exact-head runtime qualification. No local product build, query or benchmark was run. Maintained Spark 3.4/4.1 source coverage remains unavailable.
Performance
The baseline and map-coverage requests are addressed. The comment now identifies Comet JVM shuffle as the default auto/nested-off comparison. The replacement table contains all four arms and consistent JVM/native ratios from one reported cohort. The singleton control remains alongside larger maps in both input orders.
The new results retain the slow cases: arrays of structs lose to JVM shuffle at both partition counts, and the 2–50-entry reversed map at 201 partitions reports 2,449 ms native versus 2,107 ms JVM. These remain author measurements, including the disclosed Arrow 59.2.0 workaround outside the PR. I did not inspect the local raw validation bundle or independently verify its native-library identity. The description correctly limits these sequential, complete-operation timings and does not infer isolated sorting cost or a general map benefit.
Design
The version skip, shared native configuration and untimed plan assertion make case labels explicit without changing production admission rules. The --nested-hash-only entry point reuses the same runner as the full benchmark. Input preparation and plan reporting remain outside the measurement, while each timed action retains query planning, scan, key construction, shuffle and sink consumption.
Abstraction & complexity
The small recursive type predicate and expanded shape list fit the existing helper. The map expressions share the index sequence for keys and values, and the common runner keeps shape and partition coverage aligned across arms. No additional framework or production abstraction was introduced, and I have no further change to request.
andygrove
left a comment
There was a problem hiding this comment.
Both points are addressed, and the second one changed the PR's conclusion, which is the best outcome a benchmark review can have.
The four-arm table with a JVM/native ratio per shape and partition count answers the question I was actually asking: under the default auto mode with nested hashing off, the shuffle lands on CometColumnarShuffle rather than Spark, so Comet JVM shuffle is what the config's default trades against. The doc comment says that now instead of naming Spark.
The map result is the valuable part. Adding 2–10 and 2–50 entries per row in forward and reversed order shows the singleton-map shape was measuring almost nothing: struct<map<string, int>, int> with 2–50 entries reversed at 201 partitions comes back 0.86x, native 2,449 ms against JVM shuffle's 2,107 ms, and the array-of-struct cases are also slower under native in this run. Dropping the "maps are unaffected" claim and recording the regression instead is the right call, and keeping the singleton case as a control makes the contrast legible rather than just deleting the misleading row.
Approving the benchmark. What I would want said out loud before anyone reads these numbers as a case for flipping spark.comet.shuffle.native.partitioning.hash.nested.enabled on by default: the table now contains shapes where the default is faster than the feature. Two of them are the wide-map cases and the rest are array-of-struct. Whatever the eventual default, this PR has made the argument for it checkable, which is what it was for.
One follow-up rather than a change here. You note these are end-to-end sequential measurements that do not isolate sorting from hashing, and the reversed-order rows moving so much against forward order suggests the sort is carrying a lot of it. A criterion bench over the hash kernel alone, at the same entry counts, would say whether the wide-map regression is in hash_funcs/utils.rs or in the sort that Spark's InsertMapSortInRepartitionExpressions inserts ahead of it. That would decide whether there is anything to fix in the kernel at all. Worth an issue if you agree.
|
Thanks @sunchao @andygrove |
Which issue does this PR close?
Closes #5787.
Rationale for this change
Native shuffle can admit nested hash partitioning keys with
spark.comet.shuffle.native.partitioning.hash.nested.enabled(#5567), but the existing shufflebenchmarks do not measure that path. The array/struct benchmarks lack a native arm, and the
deeply nested benchmark uses keyless, round-robin repartitioning. This is separate from
#2904, which measured columnar/JVM shuffle throughput with keyless repartitioning.
With the default shuffle mode
auto, disabling nested native hashing selects Comet JVMshuffle for these shapes. The relevant configuration comparison is therefore Comet JVM
versus Comet native shuffle. The all-Spark arm also changes scan and projection execution;
Comet with Spark shuffle represents the fallback under explicitly selected
nativemode.What changes are included in this PR?
Adds a shared nested-key benchmark with Spark, Comet Spark shuffle, Comet JVM shuffle and
Comet native shuffle arms, at 5 and 201 partitions. Coverage includes a simple struct,
a primitive array, a struct containing a primitive array, an array of structs, and a struct
containing a string-to-int map. The map cases retain a singleton control and add variable
2–10 and 2–50 entry maps in forward and reversed input order. Each map has distinct keys;
reversing the order preserves its key/value pairs.
The native arm explicitly enables nested hashing. Spark versions below 4.0 skip native
map-containing keys because they do not insert the required map normalization. For admitted
cases, an untimed physical-plan check requires exactly one
CometNativeShuffleexchangeand records the plan before benchmarking, so unexpected fallback fails rather than being
reported under a native label. Map plans on Spark 4.1 show the inserted
mapsortexpression.--nested-hash-onlyruns this section directly without a temporary driver.How are these changes tested?
The following table measures the benchmark source in
8d9eede01in one complete run on2026-09-09: Apple M4 Max, Spark 4.1.3,
Zulu JDK 17.0.16,
local[5], 1,048,576 rows and a 20 GiB JVM heap. Times are best millisecondsfrom the existing harness (at least five iterations, 500 ms warmup and measurement floors).
JVM/nativeis Comet JVM time divided by Comet native time; values above 1 favor native.All four columns come from this run. No earlier measurements are mixed in.
Input-file preparation is outside the timer; planning, scan, key construction, shuffle and
no-op sink consumption are timed. Results therefore characterize these complete operations,
not isolated hashing or sorting cost. Shapes run sequentially on a local machine, and some
cases have substantial variance; the raw output retains averages and standard deviations.
The native library was rebuilt in release mode and matched by SHA-256 to the library copied
into the JVM classes. Current worktree classes were used for both Comet modules. The local
Arrow 59.2.0 lockfile workaround was used for this run and is excluded from the PR.
struct<int, string>struct<int, string>array<int>array<int>struct<array<int>, string>struct<array<int>, string>array<struct<int, string>>array<struct<int, string>>struct<map<string, int>, int>struct<map<string, int>, int>struct<map<string, int>, int> (2-10 entries, forward)struct<map<string, int>, int> (2-10 entries, forward)struct<map<string, int>, int> (2-10 entries, reversed)struct<map<string, int>, int> (2-10 entries, reversed)struct<map<string, int>, int> (2-50 entries, forward)struct<map<string, int>, int> (2-50 entries, forward)struct<map<string, int>, int> (2-50 entries, reversed)struct<map<string, int>, int> (2-50 entries, reversed)The array-of-struct cases are slower under native than JVM shuffle in this run. The larger
map cases also show that the singleton control does not establish a general map benefit:
with 2–50 entries in reversed input order at 201 partitions, native is 2,449 ms versus
2,107 ms for JVM shuffle. Differences between forward and reversed rows can also reflect machine load and
JIT/GC effects within this sequential run; they do not isolate sorting cost.
Validation:
test-compile, Spotless and Scalastyle passed.with 18 native exchange checks recorded.
skips at both 5 and 201 partitions (14 measured arms). Its timings are not in this table.
Raw build logs, physical plans, benchmark output and source/native checksums are retained
in the local validation bundle.
Additional context
This benchmark remains independent of the kernel optimization in #5778, tracked by #5777.
Native admission is checked directly in the harness; timing differences alone are not
used as evidence of native execution.