andygrove commented on code in PR #5723:
URL: https://github.com/apache/datafusion-comet/pull/5723#discussion_r3997545947
##########
spark/src/test/scala/org/apache/comet/exec/CometAggregateSuite.scala:
##########
@@ -1566,6 +1566,108 @@ class CometAggregateSuite extends CometTestBase with
AdaptiveSparkPlanHelper {
}
}
+ test("skip partial aggregation preserves post-shuffle distinct") {
Review Comment:
This test compares results but does not confirm that the stage 1 partial
actually bypassed aggregation. If the probe never fires, the test passes
without exercising the DISTINCT dedup path it is meant to protect. It might be
worth collecting the `CometHashAggregateExec` nodes and asserting
`skipped_aggregation_rows` is greater than zero on the Partial stage, the same
way the eligibility test does.
##########
spark/src/main/scala/org/apache/spark/sql/comet/operators.scala:
##########
@@ -1734,6 +1734,15 @@ trait CometBaseAggregate {
if (aggregateExpressions.isEmpty) {
val hashAggBuilder = OperatorOuterClass.HashAggregate.newBuilder()
hashAggBuilder.addAllGroupingExprs(groupingExprs.map(_.get).asJava)
+ // Spark has no expression mode to serialize here. An empty aggregate
with a required child
+ // distribution must fully deduplicate its keys (Final, or a
pre-distinct PartialMerge), so
Review Comment:
This comment says a distribution-required stage must fully deduplicate its
keys, including the pre-distinct `PartialMerge` case. For the non-empty
`PartialMerge` stage, Comet still plans a DataFusion `Partial` with
`MergeAsPartial` accumulators, and DataFusion's Partial mode uses `EmitEarly`
under memory pressure, which can re-emit a key that was already emitted. Does
that mean stage 3's distinct `count(k)` can overcount when stage 2 spills? I
think this is pre-existing and separate from this PR, and the guard correctly
refuses to skip in that shape, but could you open a tracking issue and link it
from this comment so the obligation is recorded?
##########
docs/source/user-guide/latest/tuning.md:
##########
@@ -31,6 +31,46 @@ It is recommended that `COMET_WORKER_THREADS` be set to the
number of executor c
in some environments, such as Kubernetes, where the number of cores allocated
to a pod will already be equal to the
number of executor cores.
+## Adaptive Partial Aggregation
+
+For high-cardinality grouping, Comet can bypass partial hash aggregation when
it is not
+reducing the number of rows enough. This currently applies only to fused
native shuffle-writer
+plans whose partial aggregates are grouping-only or single-argument `COUNT`.
Low-cardinality
+inputs continue to aggregate normally. The SQL metric `rows bypassing partial
aggregation`
+shows whether skipping occurred.
+
+Eligibility is conservative for the whole fused native plan: any unsupported
partial accumulator,
+Spark `PartialMerge`, or mixed-mode aggregate disables skipping in that plan.
Multi-argument
+`COUNT` and other accumulators are not admitted. Distribution-required
grouping-only stages
+still fully deduplicate, and non-native-shuffle plans retain ordinary
aggregation.
+The DataFusion testing configuration override does not bypass these safety
checks.
+
+DataFusion 55 defaults to probing after 100,000 input rows per partial
aggregation
+partition and skipping when the number of groups divided by input rows exceeds
`0.8`.
+To experiment with these thresholds, enable
`spark.comet.exec.respectDataFusionConfigs`,
+a development and testing option that defaults to `false`. For example, the
following
+SQL settings pass through the default threshold values, which you can adjust:
+
+```sql
+SET spark.comet.exec.respectDataFusionConfigs=true;
+SET
spark.comet.datafusion.execution.skip_partial_aggregation_probe_rows_threshold=100000;
+SET
spark.comet.datafusion.execution.skip_partial_aggregation_probe_ratio_threshold=0.8;
+```
+
+A lower row threshold allows an earlier decision; a lower ratio threshold makes
+skipping more likely. Skipping can increase the number of partial states
emitted
+and the amount of shuffle data, so measure the effect on your workload.
+
+To disable skipping, keep `spark.comet.exec.respectDataFusionConfigs=true` and
set
Review Comment:
The only documented way to turn this off is
`spark.comet.exec.respectDataFusionConfigs`, which is a TESTING category
option, plus a raw DataFusion key. Since this feature was disabled once before
for a TPC-DS wrong-result (#788, apache/datafusion#11850) and is now on by
default, would you consider a first-class boolean such as
`spark.comet.exec.aggregate.skipPartial.enabled` (default true) that the guard
in `configure_skip_partial_aggregation` checks alongside the plan walk? That
gives operators a supported kill switch without pointing them at a testing flag.
--
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]
