Columnar shuffle with nested types is slower than Spark

[andygrove]

Describe the bug

Based on the new benchmarks in #2902, it seems that Comet columnar shuffle with complex types is extremely slow.

Shuffle with nested schema (maxDepth=2, partitionNum=201):  Best Time(ms)   Avg Time(ms)   Stdev(ms)    Rate(M/s)   Per Row(ns)   Relative
-----------------------------------------------------------------------------------------------------------------------------------------
Spark                                                                 85             93           4          0.0       84642.1       1.0X
Comet (Spark Shuffle)                                                 84             92          11          0.0       83718.9       1.0X
Comet (jvm Shuffle)                                                  836            838           3          0.0      836109.3       0.1X
Comet (native Shuffle)                                                83             88           6          0.0       82666.6       1.0X

Steps to reproduce

SPARK_GENERATE_BENCHMARK_FILES=1 make benchmark-org.apache.spark.sql.benchmark.CometShuffleBenchmark

Expected behavior

No response

Additional context

No response

[andygrove]

performance degrades as partitions increase

OpenJDK 64-Bit Server VM 17.0.17+10-Ubuntu-122.04 on Linux 6.8.0-90-generic
AMD Ryzen 9 7950X3D 16-Core Processor
Shuffle with nested schema (maxDepth=2, partitionNum=5):  Best Time(ms)   Avg Time(ms)   Stdev(ms)    Rate(M/s)   Per Row(ns)   Relative
---------------------------------------------------------------------------------------------------------------------------------------
Spark                                                               72            103          36          0.0       72458.8       1.0X
Comet (Spark Shuffle)                                               64             69           4          0.0       63977.1       1.1X
Comet (jvm Shuffle)                                                 86             91           5          0.0       86170.0       0.8X
Comet (native Shuffle)                                              60             65           6          0.0       60448.2       1.2X

OpenJDK 64-Bit Server VM 17.0.17+10-Ubuntu-122.04 on Linux 6.8.0-90-generic
AMD Ryzen 9 7950X3D 16-Core Processor
Shuffle with nested schema (maxDepth=2, partitionNum=201):  Best Time(ms)   Avg Time(ms)   Stdev(ms)    Rate(M/s)   Per Row(ns)   Relative
-----------------------------------------------------------------------------------------------------------------------------------------
Spark                                                                 87             95           6          0.0       86502.3       1.0X
Comet (Spark Shuffle)                                                 85             89           3          0.0       84949.9       1.0X
Comet (jvm Shuffle)                                                  875            890          22          0.0      874661.3       0.1X
Comet (native Shuffle)                                                85             92          12          0.0       84685.0       1.0X

[andygrove]

With small number of partitions, hash-based shuffle is used.

With large number of partitions, sort-based shuffle is used.

[andygrove]

It turns out that the current microbenchmark is very misleading and is testing a pathological case for sort-based shuffle with a single row per output partition. It may be better to rely on TPC-* for benchmarking shuffle.