mixermt commented on issue #4859: URL: https://github.com/apache/datafusion-comet/issues/4859#issuecomment-5237171655
Hi @mbutrovich @adriangb I’ve tested nested reads, and it appears that pruning works effectively. Additionally, I’ve compared the output data to ensure complete correctness. Input bytes were even lower than what was in Spark (at least by what was reported in Spark UI and metrics) Unfortunately execution was still slower with Comet (maybe worth for investigation in other issue) Report was generated by Codex 5.6 Sol Ultra ### Schema comparison | Layer | Baseline SQL 13 | Comet SQL 6 | Result | |---|---:|---:|---| | Source path | Same 10,500 files, 2.825 TiB | Same | Identical input | | Source schema | 602 top-level fields, 2,609 primitive leaves | Same | Sampled source footers identical | | Applied scan schema | 4 top-level fields, 13 leaves | 4 top-level fields, 13 leaves | Identical | | Pushed filters | Same two filters | Same two filters | Identical | | Scan output | 371,800,464 rows | 371,800,464 rows | Identical | | Final writer schema | Seven fields | Seven fields | Identical | ### Overall comparison Metric | Baseline SQL 13 | Comet SQL 6 | Comet change -- | -- | -- | -- SQL wall time | 69.724 s | 125.843 s | +80.5% Summed executor run time | 76.476 task-min | 111.609 task-min | +45.9% Spark-reported executor CPU | 62.869 CPU-min | 33.940 CPU-min | −46.0%* Actual pod CPU, Thanos/cAdvisor | ≈63.49 CPU-min | ≈107.96 CPU-min | ≈+70.1% Average CPU cores during SQL | ≈54.6 | ≈51.5 | −5.8% JVM GC task time | 55.144 s | 61.093 s | +10.8% GC / executor run | 1.20% | 0.91% | lower share Input records | 371,800,464 | 371,800,464 | identical Reported input bytes | 51.636 GiB | 15.612 GiB | −69.8%** Shuffle bytes | 3.890 GiB | 6.981 GiB | +79.5% Shuffle records | 199,926,600 | 205,748,931 | +2.91% Final output | 4.292 MiB / 269,461 rows | 4.591 MiB / 269,461 rows | same rows, +6.97% bytes Output files | 71 | 134 | +88.7% Spill / failed tasks / retries | 0 / 0 / 0 | 0 / 0 / 0 | equal `*` Spark’s task-thread CPU counter misses much of Comet’s native worker CPU. Historical container counters show that Comet consumed more CPU, not less. `**` Native Comet and Spark input-byte counters have different semantics. Pod network receive was approximately 127.3 GiB baseline vs 91.8 GiB Comet, only about 28% lower overall. ### Where the time was spent Metric | Baseline SQL 13 | Comet SQL 6 | Comet change -- | -- | -- | -- SQL wall time | 69.724 s | 125.843 s | +80.5% Summed executor run time | 76.476 task-min | 111.609 task-min | +45.9% Spark-reported executor CPU | 62.869 CPU-min | 33.940 CPU-min | −46.0%* Actual pod CPU, Thanos/cAdvisor | ≈63.49 CPU-min | ≈107.96 CPU-min | ≈+70.1% Average CPU cores during SQL | ≈54.6 | ≈51.5 | −5.8% JVM GC task time | 55.144 s | 61.093 s | +10.8% GC / executor run | 1.20% | 0.91% | lower share Input records | 371,800,464 | 371,800,464 | identical Reported input bytes | 51.636 GiB | 15.612 GiB | −69.8%** Shuffle bytes | 3.890 GiB | 6.981 GiB | +79.5% Shuffle records | 199,926,600 | 205,748,931 | +2.91% Final output | 4.292 MiB / 269,461 rows | 4.591 MiB / 269,461 rows | same rows, +6.97% bytes Output files | 71 | 134 | +88.7% Spill / failed tasks / retries | 0 / 0 / 0 | 0 / 0 / 0 | equal The reduce/write stage alone adds about 31.35 seconds, roughly 56% of the total wall-time gap. The native shuffle writes only 2.9% more records, but 79.5% more bytes: Baseline: 20.9 shuffle bytes/record Comet: 36.4 shuffle bytes/record Difference: +74.4% per record Comet also reports 14.8 task-minutes of native shuffle decoding/decompression. The final tasks have comparable partition sizes, yet median task duration rises from 2.656 s to 17.505 s. There is no meaningful reducer skew; the slowdown is broad across tasks. ### Plan comparison ``` Baseline: Parquet scan → ColumnarToRow → Spark Filter/Project/Generate ×2 → HashAggregate → Spark Exchange → AQE (71 reducers) → HashAggregate → Parquet writer Comet: CometNativeScan → Comet Filter/Project/Explode ×2 → CometHashAggregate → CometExchange / CometNativeShuffle → AQE (134 reducers) → CometHashAggregate → CometColumnarToRow → Spark Parquet writer ``` Driver logs say Comet accelerated 13 of 15 eligible operators (86%). The only substantive fallback is DataWritingCommandExec, because spark.comet.parquet.write.enabled=false Baseline executors were 6/7 in one DC. Comet executors were spread across multiple DCs. This might contribute to the much longer remote shuffle requests but usually we don't see any CPU resources were equal: seven executors, 12 cores each, 9-core request/12-core limit, 84 task slots. -- 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]
