Comet Native Executors: TPC-H q17 in pipelined mode takes 277 s where Ballista takes 154 s #7409

What is the problem the feature request solves?

TPC-H q17 at SF1000 in pipelined mode runs in about 277 s on Comet Native Executors (#7289, head e9d6e3da0) and 154 s on Ballista, on the same cluster: 2 nodes linked by 2.5 GbE, one executor per node with 8 cores and 56 GiB, node-local Parquet.

Both engines run the same plan shape. Neither broadcasts part in pipelined mode, so both shuffle all of lineitem twice: once to join part, and once for the per-l_partkey average. Both move about 35 GB per node in each direction.

Sampling CPU and network on both nodes every 5 seconds shows where the time goes:

• Ballista keeps the network at about 227 MB/s each way, close to the link's limit, for most of the query, with the 8 executor cores never all busy. It is network-bound.
• Comet averages 134 MB/s. It has a phase of about 90 s with all 8 executor cores busy and the network nearly idle, and an idle stretch of about 15 s before any data moves. With partial aggregation skipping, the busy phase is about 60 s.

Describe the potential solution

In order of expected gain:

1. Comet Native Executors: broadcast a small build side in pipelined mode #7408: broadcast a small build side in pipelined mode. This removes one of the two lineitem shuffles, about half of the network traffic, which bounds both engines here.
2. Comet Native Executors: let a fragment's partial aggregate stop aggregating, including AVG #7406: let a fragment's partial aggregate stop aggregating, including AVG. Measured at -13 to -18%, and it removes the OOM.
3. Comet Native Executors: choose partition counts for pipelined mode #7407: choose partition counts for pipelined mode from the cluster's size. Spark's 1,792 scan partitions and 200 shuffle partitions all run at once in pipelined mode.
4. Profile the remaining CPU-bound phase on an executor. It is likely the final aggregate over 200 million groups or the join stage, but that is not measured yet.
5. Find out what the first 15 s of a Comet query wait on. Ballista starts moving data at once.

Additional context

The blocking-mode side of q17 is #7391. Part of #7288.