QinXi-ai commented on code in PR #25988:
URL: https://github.com/apache/datafusion/pull/25988#discussion_r4500464729
##########
datafusion/optimizer/src/single_distinct_to_groupby.rs:
##########
@@ -154,12 +155,41 @@ fn is_single_distinct_agg(
if aggregate_count != aggr_expr.len() || fields_set.len() != 1 {
return Ok(false);
}
+ // With one target partition, a native GroupsAccumulator avoids building an
+ // extra aggregate with a row per (group, distinct value) pair. With
multiple
+ // partitions, the rewrite also distributes deduplication by both keys;
+ // keeping DISTINCT instead repartitions lists of values by group and
merges
+ // them in the final accumulator. That can concentrate work on a few
groups.
+ // Native accumulator support alone does not establish that this is
cheaper.
+ if use_native_grouped_distinct
Review Comment:
Thanks, I missed this because the spill test only had one distinct value per
group. I've made the native path opt-in with `single_distinct_prefer_native`,
defaulting to `false`. Your 1 MiB reproducer now passes with the default; the
test also checks the native EXPLAIN and memory error with the flag on. I used
your `prefer_native` name since native can use more memory too. Agreed on
sharing one setting with #26000.
##########
benchmarks/sql_benchmarks/grouped_count_distinct/init/load.sql:
##########
@@ -0,0 +1,11 @@
+CREATE TABLE grouped_distinct AS
+SELECT value % 2000 AS g,
+ value % 8 AS few_g,
+ (value * 48271) % 999983 AS x,
+ (value / 2000) % 8 AS low_x,
+ arrow_cast((value * 48271) % 999983, 'UInt64') AS unsigned_x,
+ CAST((value * 48271) % 999983 AS VARCHAR) AS string_x,
+ CAST((value * 48271) % 999983 AS DOUBLE) AS float_x,
+ CASE WHEN value % 100 < 90 THEN 0 ELSE value % 8 END AS skew_g,
+ CAST(CASE WHEN value % 100 < 90 THEN 0 ELSE value % 8 END AS VARCHAR)
AS string_g
Review Comment:
Added Q10 with those columns and reproduced the slowdown: at 4M rows and one
target partition, I got 398.12 ms with native enabled vs 284.87 ms with it off,
about 40% slower (15 retained samples each). I've put this alongside the
original workload's improvement in the description and added off/on commands to
the suite README.
##########
datafusion/optimizer/src/single_distinct_to_groupby.rs:
##########
@@ -231,6 +261,8 @@ impl OptimizerRule for SingleDistinctToGroupBy {
&aggr_expr,
input.schema(),
count_rollup.as_ref(),
+ !group_expr.is_empty()
+ && config.options().execution.target_partitions == 1,
Review Comment:
You're right, one target partition can still merge several input partitions.
Added your UNION ALL query and a table built at four partitions then queried at
one. Both have result checks and physical-plan snapshots showing
Partial/Coalesce/Final. The comment now calls out these cases.
--
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]
