andygrove opened a new pull request, #5388: URL: https://github.com/apache/datafusion-comet/pull/5388
## Which issue does this PR close? Part of https://github.com/apache/datafusion-comet/issues/5373. Addresses the shuffle suite timeout raised in review of https://github.com/apache/datafusion-comet/pull/5380. ## Rationale for this change `CometShuffleBenchmark` never finishes. In the baseline run recorded in `benchmarks/results/micro/RUN-INFO.md` it hit the runner's one hour timeout partway through the struct group, so the shuffle family has no published results and the truncated artifact was dropped from #5380. Raising the timeout does not fix it. Two independent defects prevent the suite from completing at all, and separately the suite is far slower than a microbenchmark should be. Two blockers: 1. The struct and dictionary groups ran at `1024 * 1024 * 100` rows while the equivalent array group ran at `1024 * 1024 * 1`. Both select a constant (`CAST(1 AS <type>)`), so every row is identical and repartitioning on that column routes everything to a single partition. The extra rows bought no coverage, only runtime: each struct case measured about 25s per implementation, putting that one group well past half an hour. 2. `shuffleDictionaryBenchmark` built its data with `REPEAT(CAST(1 AS BINARY), 100)`. ANSI mode, the default on Spark 4.x, rejects an INT to BINARY cast: ``` [DATATYPE_MISMATCH.CAST_WITH_CONF_SUGGESTION] Cannot resolve "CAST(1 AS BINARY)" due to data type mismatch: cannot cast "INT" to "BINARY" with ANSI mode on. ``` This fails on the group's first case regardless of row count or timeout. The recorded run never reached it because it timed out during struct, so the timeout was masking a hard failure. On runtime, the dominant cost was not the work. Spark's `Benchmark` spends two seconds warming up and two seconds measuring every case, and the minimum measured duration across 576 runs was exactly 2000ms, so those floors set the suite's runtime rather than the shuffles themselves. ## What changes are included in this PR? Correctness: - Build the dictionary group's value as a string and cast to the target type, so `BINARY` works under ANSI. Runtime, measured locally from 54m48s to 8m10s: - Use 500ms warmup and measurement budgets with a higher iteration minimum, via a `microBenchmark` helper. - Drop all groups to `1024 * 1024 * 1` rows. - Cover five representative types instead of nine, hoisted into a shared `benchmarkTypes`. The shuffle paths do not branch per numeric width, so the extra integer and float widths multiplied runtime without distinguishing implementations. - Remove the single column `Shuffle` and `Shuffle with random values` groups and the now unused `shuffleBenchmark`. A single column table is not representative of real shuffles, and the wide 10 and 20 column groups already cover the same paths with realistic row widths. Partition counts are unchanged, and the high value is now documented as load bearing: it must stay above Spark's `spark.shuffle.sort.bypassMergeThreshold`. Below that threshold the write path switches to `CometBypassMergeSortShuffleWriter`, which holds a page per partition from a JVM wide pool. Lowering it to 50 raised the required shuffle memory from 8GB to roughly 16GB, which does not fit the m7i.2xlarge used for baselines, and it also hid the JVM shuffle's degradation at high fan-out. ## How are these changes tested? Ran the suite locally on Spark 4.1 and compared against the previous configuration. - Completes in 8m10s, down from 54m48s, with all 15 groups present in the results file for the first time. - Relative Spark versus Comet ratios are preserved where they matter. The wide groups at the high partition count still show Comet JVM shuffle at 0.5X to 0.7X, retaining the high fan-out degradation signal. Array, struct and dictionary ratios are unchanged within noise against the previous run. - Measurement noise: stdev as a fraction of best time is p50 6.8%, p90 13.8%. Verifying the reduced row counts required working around a separate pre-existing issue: `getSparkSession` sets `spark.executor.memoryOverhead`, which is Spark's overhead rather than Comet's, so the JVM shuffle pool stays at its default and the first nested case fails with `UNABLE_TO_ACQUIRE_MEMORY` on a many core machine. Runs here passed `-Dspark.comet.memoryOverhead=8g`. That is left for a follow up along with the redundant per case table writes. -- 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]
