[VL] The time taken to merge payload during the shuffle write is excessively high

[NEUpanning]

Backend

VL (Velox)

Bug description

The total time across all tasks for vanilla Spark was 78.1 hours, but for gluten it reached 1899.3 hours. The flame graph shows that the majority of time is occupied by the merging of payloads. After adding some logs, I see merge operation occurred 1522330 times for 1084128 rows in a task, with each instance taking a few milliseconds.

Flame graph:

gluten shuffle metrics:

shuffle records written: 39,407,858,231
shuffle write time total (min, med, max (stageId: taskId))
17.86 h (0 ms, 27.4 s, 17.1 m (stage 0.0: task 499))
time to compress total (min, med, max (stageId: taskId))
31.52 h (0 ms, 29.6 s, 20.3 m (stage 0.0: task 1191))
time to split total (min, med, max (stageId: taskId))
1478.09 h (0 ms, 45.5 m, 1.71 h (stage 0.0: task 86))
time to spill total (min, med, max (stageId: taskId))
15.77 h (0 ms, 25.2 s, 16.3 m (stage 0.0: task 499))

shuffle schema:

Gluten version

Gluten-1.3

Spark version

Spark-3.5.x

Spark configurations

No response

System information

No response

Relevant logs

[NEUpanning]

cc @marin-ma

[marin-ma]

Could you share other metrics for the ColumnarShuffleExchange operator? You may also try enabling sort-based shuffle and see if it can get better performance.

[NEUpanning]

@marin-ma

Could you share other metrics for the ColumnarShuffleExchange operator?

ColumnarExchange

shuffle records written: 39,407,858,231
shuffle write time total (min, med, max (stageId: taskId))
17.86 h (0 ms, 27.4 s, 17.1 m (stage 0.0: task 499))
time to compress total (min, med, max (stageId: taskId))
31.52 h (0 ms, 29.6 s, 20.3 m (stage 0.0: task 1191))
time to split total (min, med, max (stageId: taskId))
1478.09 h (0 ms, 45.5 m, 1.71 h (stage 0.0: task 86))
shuffle wall time total (min, med, max (stageId: taskId))
1527.48 h (0 ms, 46.5 m, 1.92 h (stage 0.0: task 86))
number of input rows: 39,407,858,231
time to spill total (min, med, max (stageId: taskId))
15.77 h (0 ms, 25.2 s, 16.3 m (stage 0.0: task 499))
shuffle bytes spilled total (min, med, max (stageId: taskId))
3.0 TiB (0.0 B, 1516.5 MiB, 1944.8 MiB (stage 0.0: task 909))
number of input batches: 9,788,389
Native.mergeSpillsTime total (min, med, max (stageId: taskId))
34.92 h (0 ms, 33.8 s, 23.1 m (stage 0.0: task 1191))
data size total (min, med, max (stageId: taskId))
75.6 TiB (0.0 B, 38.8 GiB, 38.9 GiB (stage 0.0: task 1))
peak bytes allocated total (min, med, max (stageId: taskId))
7.5 TiB (0.0 B, 5.1 GiB, 5.6 GiB (stage 0.0: task 1418))
number of partitions: 2,000
shuffle bytes written total (min, med, max (stageId: taskId))
3.7 TiB (0.0 B, 1944.5 MiB, 1953.5 MiB (stage 0.0: task 1))

You may also try enabling sort-based shuffle and see if it can get better performance.

The number of shuffle partitions is only 2000, which is not greater than spark.gluten.sql.columnar.shuffle.sort.partitions.threshold. I'll try to force enable it.

And I adjusted merge buffer size from 4096 to 2048, resulting in a ~30-fold reduction in split time.

[NEUpanning]

Additionally, I am wondering if we could increase the capacity of cached merge payload buffers in advance to avoid resizing them multiple times.

[marin-ma]

@NEUpanning spark.gluten.sql.columnar.shuffle.merge.threshold default 0.25, meaning any buffer < 0.25 * 4096 will be saved for merging. You may set it to a lower value or set to 0 to disable merging them.