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. High-cardinality partial aggregation can spend substantial CPU building a hash table while emitting almost as many rows as it consumes.
Release-mode Spark A/B measurements on Comet 75fdddc9285ec61c0cd326977c61dd41fca39a8b (DataFusion 55.0.0, merged in #5262) confirm the opportunity:
| Fixture |
Disabled (1.1), median wall |
Enabled (0.8), median wall |
Result |
| High-cardinality COUNT |
2,060 ms |
1,380 ms |
33.0% faster |
| High-cardinality COUNT, fresh JVM/reversed initial order |
1,501 ms |
1,027 ms |
31.6% faster |
| Low-cardinality control |
292 ms |
267 ms |
Mixed pairs/warmup effects; not a claimed gain |
| Low-cardinality control, fresh JVM/reversed initial order |
1,370 ms |
1,360 ms |
Effectively neutral under contention |
Each JVM used 16,777,216 Parquet rows, 8 writers, 32 reducers, local[8], a 10 GiB heap, Spark 4.1.3, AQE/exchange reuse disabled, native direct-read shuffle, one warmup per mode, and seven alternating measured pairs. The high-cardinality query groups by a unique integer key, counts each group, then sums the counts; the control groups into 1,024 keys. All results were checked exactly. Skipping won all 14 high-cardinality pairs. Map-stage medians were 1,248 -> 569 ms and 917 -> 413 ms. These are workstation measurements with concurrent desktop activity, not a controlled comparison between DataFusion versions or a universal no-regression guarantee.
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. The older DataFusion 54.1.0 fixture at 81d637b9bf40a5be6f4f0c65ad6f497b34746e69 measured 2,125 -> 1,560 ms (26.6%); the table above is the new 55.0.0 evidence.
Describe the potential solution
Implement a guarded reintroduction for explicitly supported accumulators and actual Spark aggregate modes, not a global threshold reduction.
- Preserve grouping-only DISTINCT semantics. Distribution-required grouping-only stages must fully deduplicate their keys. Serialize these empty-aggregate stages as native
Final, rather than leaving the protobuf default Partial. On unmodified 55.0.0 main, enabling skipping made both SELECT count(*) FROM (SELECT DISTINCT k ...) and SELECT count(DISTINCT k) ... return 2,400,000 instead of 1,980,000. The local mode fix restores both results at 2.4 million input rows, and its focused ScalaTest passes (1 test covering both shapes).
- Use explicit accumulator/mode eligibility. Start narrowly with grouping-only/COUNT cases. Exclude Spark
PartialMerge and mixed-mode stages even when represented as native Partial; their inputs are already aggregate states. Unsupported accumulators must retain ordinary aggregation, including when users opt into skipping. Grouped integer SUM and mixed DISTINCT/COUNT both reproduce Input batch conversion to state not implemented under unrestricted skipping on 55.0.0, even after the DISTINCT fix. Do not admit SUM, custom accumulators, or variable-width aggregate states without independent conversion, memory, and spill validation.
- Keep the supported path covered by runtime regressions. After the mode fix, nine supported SQL cases (COUNT, nullable/filtered COUNT, both DISTINCT shapes, empty grouped/global input, null grouping keys, all-null COUNT values) match Spark with skipping off/on. Together with the two unsupported cases' disabled controls this gives 20 matching results; the two enabled unsupported cases still fail as expected until eligibility is implemented. Add regressions that require those excluded cases to remain correct without skipping, and prove eligible cases actually skip.
- Validate bounded memory and controls. The same 16,777,216-row fixture with
spark.comet.memoryOverhead=128m passed all eight executions (two warmups plus three measured pairs), with 64 aggregate spills and approximately 273 MB spilled per query in both modes. Default-memory runs had zero spills. This validates one spilling workload, not all memory pools, AQE, distributed deployment, or variable-width states. Preserve low-cardinality controls and rerun the guarded implementation.
Guarded implementation validation
A local implementation on the same DataFusion 55 main base now combines the DISTINCT fix with a conservative fused-plan eligibility gate. It admits only native shuffle-writer plans whose partial aggregates are grouping-only or single-argument COUNT. Any unsupported partial accumulator or PartialMerge/mixed mode disables skipping throughout that fused plan. Non-native-shuffle roots remain disabled, and testing overrides cannot bypass eligibility. A new rows bypassing partial aggregation SQL metric verifies actual skipping.
- Native eligibility regression: 1 passed. Full Spark aggregate suite: 93 passed, zero failures, 2 ignored. The two new focused tests also passed independently.
- The 2.4-million-row correctness harness now has 22 exact matches against Spark and zero conversion errors, including both previously excluded error cases, which retain ordinary aggregation.
- Guarded high-cardinality A/B, seven alternating pairs: 664 -> 479 ms median wall time (27.9% faster); enabled won all seven pairs. Enabled queries bypassed 15,925,248 rows each; disabled queries bypassed zero.
- Low-cardinality control, seven pairs: 199 -> 185 ms; both modes bypassed zero rows. No optimization gain is claimed for this control.
- The 128 MiB native-memory run passed all eight executions (two warmups plus three pairs), with 64 aggregate spills and approximately 273 MB spilled per query. Enabled queries still bypassed 15,925,248 rows, so skipping and spilling were exercised together. Median wall times were 945 -> 858 ms, a limited performance sample.
These use the same 16,777,216-row fixture and settings described above. Compilation and test processes had finished before timing. Compare modes within each run, not absolute times across the earlier and guarded runs. The implementation is local; these results do not imply a published or merged PR, nor coverage of every memory pool, AQE, or distributed deployment.
Additional context
DataFusion 55.0.0 is already integrated, but it was not a prerequisite for COUNT skipping: 54.1.0 already supported that path. The earlier expectation of a new capability guard was incorrect. DataFusion 55 removes GroupsAccumulator::supports_convert_to_state() and makes convert_to_state() required. Comet #5262 supplies error-returning placeholders for several custom accumulators, including integer SUM and MergeAsPartialGroupsAccumulator; required trait methods do not imply working conversion support.
Skip-partial was disabled in #788 after correctness failures; apache/datafusion#11850 records the historical failure. The proposal is a guarded implementation with explicit correctness boundaries, not unrestricted enablement.
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. High-cardinality partial aggregation can spend substantial CPU building a hash table while emitting almost as many rows as it consumes.Release-mode Spark A/B measurements on Comet
75fdddc9285ec61c0cd326977c61dd41fca39a8b(DataFusion 55.0.0, merged in #5262) confirm the opportunity:1.1), median wall0.8), median wallEach JVM used 16,777,216 Parquet rows, 8 writers, 32 reducers,
local[8], a 10 GiB heap, Spark 4.1.3, AQE/exchange reuse disabled, native direct-read shuffle, one warmup per mode, and seven alternating measured pairs. The high-cardinality query groups by a unique integer key, counts each group, then sums the counts; the control groups into 1,024 keys. All results were checked exactly. Skipping won all 14 high-cardinality pairs. Map-stage medians were 1,248 -> 569 ms and 917 -> 413 ms. These are workstation measurements with concurrent desktop activity, not a controlled comparison between DataFusion versions or a universal no-regression guarantee.The experiment used
spark.comet.exec.respectDataFusionConfigs=true, toggledspark.comet.datafusion.execution.skip_partial_aggregation_probe_ratio_threshold, and heldspark.comet.datafusion.execution.coalesce_batches=falsein both modes. The older DataFusion 54.1.0 fixture at81d637b9bf40a5be6f4f0c65ad6f497b34746e69measured 2,125 -> 1,560 ms (26.6%); the table above is the new 55.0.0 evidence.Describe the potential solution
Implement a guarded reintroduction for explicitly supported accumulators and actual Spark aggregate modes, not a global threshold reduction.
Final, rather than leaving the protobuf defaultPartial. On unmodified 55.0.0 main, enabling skipping made bothSELECT count(*) FROM (SELECT DISTINCT k ...)andSELECT count(DISTINCT k) ...return 2,400,000 instead of 1,980,000. The local mode fix restores both results at 2.4 million input rows, and its focused ScalaTest passes (1 test covering both shapes).PartialMergeand mixed-mode stages even when represented as nativePartial; their inputs are already aggregate states. Unsupported accumulators must retain ordinary aggregation, including when users opt into skipping. Grouped integer SUM and mixed DISTINCT/COUNT both reproduceInput batch conversion to state not implementedunder unrestricted skipping on 55.0.0, even after the DISTINCT fix. Do not admit SUM, custom accumulators, or variable-width aggregate states without independent conversion, memory, and spill validation.spark.comet.memoryOverhead=128mpassed all eight executions (two warmups plus three measured pairs), with 64 aggregate spills and approximately 273 MB spilled per query in both modes. Default-memory runs had zero spills. This validates one spilling workload, not all memory pools, AQE, distributed deployment, or variable-width states. Preserve low-cardinality controls and rerun the guarded implementation.Guarded implementation validation
A local implementation on the same DataFusion 55 main base now combines the DISTINCT fix with a conservative fused-plan eligibility gate. It admits only native shuffle-writer plans whose partial aggregates are grouping-only or single-argument COUNT. Any unsupported partial accumulator or PartialMerge/mixed mode disables skipping throughout that fused plan. Non-native-shuffle roots remain disabled, and testing overrides cannot bypass eligibility. A new
rows bypassing partial aggregationSQL metric verifies actual skipping.These use the same 16,777,216-row fixture and settings described above. Compilation and test processes had finished before timing. Compare modes within each run, not absolute times across the earlier and guarded runs. The implementation is local; these results do not imply a published or merged PR, nor coverage of every memory pool, AQE, or distributed deployment.
Additional context
DataFusion 55.0.0 is already integrated, but it was not a prerequisite for COUNT skipping: 54.1.0 already supported that path. The earlier expectation of a new capability guard was incorrect. DataFusion 55 removes
GroupsAccumulator::supports_convert_to_state()and makesconvert_to_state()required. Comet #5262 supplies error-returning placeholders for several custom accumulators, including integer SUM andMergeAsPartialGroupsAccumulator; required trait methods do not imply working conversion support.Skip-partial was disabled in #788 after correctness failures; apache/datafusion#11850 records the historical failure. The proposal is a guarded implementation with explicit correctness boundaries, not unrestricted enablement.