peterxcli opened a new issue, #5717: URL: https://github.com/apache/datafusion-comet/issues/5717
### What is the problem the feature request solves? Comet disables DataFusion's adaptive skip-partial aggregation by setting the probe ratio threshold to `1.1`. For high-cardinality grouping, a partial hash aggregate can spend substantial CPU building a hash table while emitting almost as many rows as it consumes. A local Spark A/B at Comet commit `81d637b9bf40a5be6f4f0c65ad6f497b34746e69` used 16,777,216 Parquet rows, 8 writers, 32 reducers, `local[8]`, AQE disabled, native direct-read shuffle, one warmup per mode, and seven alternating measured runs. The query groups by a unique integer key, counts each group, then sums the counts. The control groups into 1,024 keys. Every run returned the expected total. | Median | Disabled (`1.1`) | Enabled (`0.8`) | | --- | ---: | ---: | | High-cardinality query wall time | 2,125 ms | 1,560 ms | | High-cardinality map-stage wall time | 1,315 ms | 801 ms | | High-cardinality native partial compute, summed across tasks | 4,049 ms | 91 ms | | Low-cardinality query wall time | 364 ms | 373 ms | This is a 26.6% end-to-end improvement on the high-cardinality fixture. The low-cardinality difference is small and mixed across runs; this is not evidence of a universal speedup. The experiment used `spark.comet.exec.respectDataFusionConfigs=true`, toggled `spark.comet.datafusion.execution.skip_partial_aggregation_probe_ratio_threshold`, and held `spark.comet.datafusion.execution.coalesce_batches=false` in both modes. ### Describe the potential solution Re-enable adaptive skip-partial aggregation only for eligible partial aggregates, after addressing these prerequisites: 1. Preserve the required semantics of grouping-only aggregate stages. The empty `aggregateExpressions` serialization path currently leaves the protobuf mode at its default, `Partial`, including distribution-required stages that must fully deduplicate. With skip-partial enabled, those stages can emit duplicate keys without a later deduplication step. A reproduction with eight overlapping input partitions returned 2,400,000 rather than 1,980,000 distinct keys. Serialize distribution-required grouping-only stages as native `Final`; test both `SELECT count(*) FROM (SELECT DISTINCT k ...)` and `SELECT count(DISTINCT k) ...`. 2. Gate eligibility on accumulator support and actual Spark aggregate modes. Do not enable skipping for Spark `PartialMerge` or mixed-mode stages simply because they are represented as native `Partial`. Start with supported grouping-only/COUNT cases, and validate capability handling against the DataFusion 55 upgrade (#5262); Comet's custom accumulator conversion paths need particular care. 3. Validate empty/null inputs, DISTINCT, memory limits/spilling, and low-cardinality controls before changing the default. The correctness prerequisite can be reviewed separately from default enablement. A local focused regression for both DISTINCT shapes passes with that prerequisite applied. ### Additional context Skip-partial was disabled in #788 after correctness failures; apache/datafusion#11850 records the historical failure. This issue proposes a guarded reintroduction with a concrete regression and workload evidence, rather than changing the threshold globally. -- 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]
