andygrove opened a new issue, #7409: URL: https://github.com/apache/datafusion-comet/issues/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. | | q17 | | --- | ---: | | Ballista (`ballista.execution.mode=pipelined`, 16 partitions) | 154.9, 153.9, 152.2, 157.5 s | | Comet Native Executors, default | 293.3, 276.7, 276.1 s; one executor OOM-killed on a repeated run | | Comet, `spark.sql.shuffle.partitions=16`, `spark.sql.files.maxPartitionNum=16` | 378.1, 374.7 s | | Comet, with partial aggregation skipping (experiment, #7406) | 241.8, 242.4 s | 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. #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. #7406: let a fragment's partial aggregate stop aggregating, including `AVG`. Measured at -13 to -18%, and it removes the OOM. 3. #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. -- 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]
