Comet Native Executors: choose partition counts for pipelined mode #7407

What is the problem the feature request solves?

In pipelined mode a Ballista job places every task of every stage at once, and all partitions of all stages run concurrently, sharing the cores (plan_placement in Ballista's pipelined_graph.rs). Comet Native Executors take their partition counts from Spark's plan, which is sized for stage-by-stage execution:

• TPC-H q17 at SF1000 reads lineitem as 1,792 partitions in each of two scan stages, and shuffles into 200 partitions (spark.sql.shuffle.partitions). That is about 4,000 partition streams on 16 cores, and two streaming 1,792 x 200 shuffles.
• An executor of 56 GiB with an 18.7 GB memory pool was OOM-killed at 58.6 GB resident on a repeated q17 run. Most of that memory is outside the pool.
• Asking Spark for Ballista's counts instead, spark.sql.shuffle.partitions=16 and spark.sql.files.maxPartitionNum=16, gave 17 lineitem partitions for 16 cores. The 17th runs after the rest, and q17 went from about 277 s to 378 s and 375 s.

Ballista itself plans q17 with 16 partitions in every stage and runs it in 154 s on the same cluster.

Describe the potential solution

For pipelined mode, choose the partition counts from the cluster's size rather than from Spark's defaults:

• The driver groups each scan's file splits into exactly a multiple of the cluster's vcores, rather than relying on spark.sql.files.maxPartitionNum, which is only a target.
• Shuffles get a matching number of partitions.

The scheduler knows the cluster's vcores, so it could report them to the driver, or the driver could take them from a setting.

Additional context

Found while comparing q17 with Ballista (#7409). Part of #7288.