adriangb commented on code in PR #25288:
URL: https://github.com/apache/datafusion/pull/25288#discussion_r4006040186
##########
datafusion/functions-aggregate/src/approx_median.rs:
##########
@@ -147,4 +147,10 @@ impl AggregateUDFImpl for ApproxMedian {
fn documentation(&self) -> Option<&Documentation> {
self.doc()
}
+
+ // Left at the default `Honored`. The accumulator rejects `DISTINCT` with
+ // `not_impl_err!`, so `f(DISTINCT x)` only works when
+ // `SingleDistinctToGroupBy` rewrites the node and deduplicates the input
+ // first. When it cannot, for example beside an `avg`, the query errors
+ // rather than returning the non-distinct answer.
Review Comment:
This accumulator rejects `DISTINCT` with `not_impl_err!`, the same as
`approx_percentile_cont_with_weight`, which is tagged `Unsupported`. With the
variant definitions suggested in `udaf.rs`, this one is `Unsupported` too. The
full path avoids a new import outside the diff; a `use` line is better.
```suggestion
fn distinct_handling(&self) -> datafusion_expr::DistinctHandling {
// The accumulator rejects `DISTINCT` with `not_impl_err!`, so the
// planner has to deduplicate the input first.
datafusion_expr::DistinctHandling::Unsupported
}
```
##########
datafusion/sqllogictest/test_files/aggregates_simplify.slt:
##########
@@ -356,3 +356,224 @@ DROP TABLE IF EXISTS tbl;
statement ok
DROP TABLE sum_simplify_t;
+
+#######
+# EliminateAggregateDistinct: DISTINCT is dropped from duplicate-insensitive
+# aggregates, so no inner group by is planned to deduplicate the input.
+#######
+
+statement ok
+CREATE TABLE distinct_simplify_t (g INT, v INT, b BOOLEAN) AS VALUES
+ (1, 3, true),
+ (1, 3, true),
+ (1, 5, false),
+ (2, 7, true),
+ (2, 7, true),
+ (2, NULL, NULL);
+
+# min: DISTINCT cannot change the minimum
+query II
+SELECT g, min(DISTINCT v) FROM distinct_simplify_t GROUP BY g ORDER BY g;
+----
+1 3
+2 7
+
+query TT
+EXPLAIN SELECT g, min(DISTINCT v) FROM distinct_simplify_t GROUP BY g;
+----
+logical_plan
+01)Aggregate: groupBy=[[distinct_simplify_t.g]],
aggr=[[min(distinct_simplify_t.v) AS min(DISTINCT distinct_simplify_t.v)]]
+02)--TableScan: distinct_simplify_t projection=[g, v]
+physical_plan
+01)AggregateExec: mode=FinalPartitioned, gby=[g@0 as g],
aggr=[min(distinct_simplify_t.v) as min(DISTINCT distinct_simplify_t.v)]
+02)--RepartitionExec: partitioning=Hash([g@0], 4), input_partitions=1
+03)----AggregateExec: mode=Partial, gby=[g@0 as g],
aggr=[min(distinct_simplify_t.v) as min(DISTINCT distinct_simplify_t.v)]
+04)------DataSourceExec: partitions=1, partition_sizes=[1]
+
+# max
+query II
+SELECT g, max(DISTINCT v) FROM distinct_simplify_t GROUP BY g ORDER BY g;
+----
+1 5
+2 7
+
+query TT
+EXPLAIN SELECT g, max(DISTINCT v) FROM distinct_simplify_t GROUP BY g;
+----
+logical_plan
+01)Aggregate: groupBy=[[distinct_simplify_t.g]],
aggr=[[max(distinct_simplify_t.v) AS max(DISTINCT distinct_simplify_t.v)]]
+02)--TableScan: distinct_simplify_t projection=[g, v]
+physical_plan
+01)AggregateExec: mode=FinalPartitioned, gby=[g@0 as g],
aggr=[max(distinct_simplify_t.v) as max(DISTINCT distinct_simplify_t.v)]
+02)--RepartitionExec: partitioning=Hash([g@0], 4), input_partitions=1
+03)----AggregateExec: mode=Partial, gby=[g@0 as g],
aggr=[max(distinct_simplify_t.v) as max(DISTINCT distinct_simplify_t.v)]
+04)------DataSourceExec: partitions=1, partition_sizes=[1]
+
+# bool_and
+query IB
+SELECT g, bool_and(DISTINCT b) FROM distinct_simplify_t GROUP BY g ORDER BY g;
+----
+1 false
+2 true
+
+query TT
+EXPLAIN SELECT g, bool_and(DISTINCT b) FROM distinct_simplify_t GROUP BY g;
+----
+logical_plan
+01)Aggregate: groupBy=[[distinct_simplify_t.g]],
aggr=[[bool_and(distinct_simplify_t.b) AS bool_and(DISTINCT
distinct_simplify_t.b)]]
+02)--TableScan: distinct_simplify_t projection=[g, b]
+physical_plan
+01)AggregateExec: mode=FinalPartitioned, gby=[g@0 as g],
aggr=[bool_and(distinct_simplify_t.b) as bool_and(DISTINCT
distinct_simplify_t.b)]
+02)--RepartitionExec: partitioning=Hash([g@0], 4), input_partitions=1
+03)----AggregateExec: mode=Partial, gby=[g@0 as g],
aggr=[bool_and(distinct_simplify_t.b) as bool_and(DISTINCT
distinct_simplify_t.b)]
+04)------DataSourceExec: partitions=1, partition_sizes=[1]
+
+# bool_or
+query IB
+SELECT g, bool_or(DISTINCT b) FROM distinct_simplify_t GROUP BY g ORDER BY g;
+----
+1 true
+2 true
+
+query TT
+EXPLAIN SELECT g, bool_or(DISTINCT b) FROM distinct_simplify_t GROUP BY g;
+----
+logical_plan
+01)Aggregate: groupBy=[[distinct_simplify_t.g]],
aggr=[[bool_or(distinct_simplify_t.b) AS bool_or(DISTINCT
distinct_simplify_t.b)]]
+02)--TableScan: distinct_simplify_t projection=[g, b]
+physical_plan
+01)AggregateExec: mode=FinalPartitioned, gby=[g@0 as g],
aggr=[bool_or(distinct_simplify_t.b) as bool_or(DISTINCT distinct_simplify_t.b)]
+02)--RepartitionExec: partitioning=Hash([g@0], 4), input_partitions=1
+03)----AggregateExec: mode=Partial, gby=[g@0 as g],
aggr=[bool_or(distinct_simplify_t.b) as bool_or(DISTINCT distinct_simplify_t.b)]
+04)------DataSourceExec: partitions=1, partition_sizes=[1]
+
+# bit_and
+query II
+SELECT g, bit_and(DISTINCT v) FROM distinct_simplify_t GROUP BY g ORDER BY g;
+----
+1 1
+2 7
+
+query TT
+EXPLAIN SELECT g, bit_and(DISTINCT v) FROM distinct_simplify_t GROUP BY g;
+----
+logical_plan
+01)Aggregate: groupBy=[[distinct_simplify_t.g]],
aggr=[[bit_and(distinct_simplify_t.v) AS bit_and(DISTINCT
distinct_simplify_t.v)]]
+02)--TableScan: distinct_simplify_t projection=[g, v]
+physical_plan
+01)AggregateExec: mode=FinalPartitioned, gby=[g@0 as g],
aggr=[bit_and(distinct_simplify_t.v) as bit_and(DISTINCT distinct_simplify_t.v)]
+02)--RepartitionExec: partitioning=Hash([g@0], 4), input_partitions=1
+03)----AggregateExec: mode=Partial, gby=[g@0 as g],
aggr=[bit_and(distinct_simplify_t.v) as bit_and(DISTINCT distinct_simplify_t.v)]
+04)------DataSourceExec: partitions=1, partition_sizes=[1]
+
+# bit_or
+query II
+SELECT g, bit_or(DISTINCT v) FROM distinct_simplify_t GROUP BY g ORDER BY g;
+----
+1 7
+2 7
+
+query TT
+EXPLAIN SELECT g, bit_or(DISTINCT v) FROM distinct_simplify_t GROUP BY g;
+----
+logical_plan
+01)Aggregate: groupBy=[[distinct_simplify_t.g]],
aggr=[[bit_or(distinct_simplify_t.v) AS bit_or(DISTINCT distinct_simplify_t.v)]]
+02)--TableScan: distinct_simplify_t projection=[g, v]
+physical_plan
+01)AggregateExec: mode=FinalPartitioned, gby=[g@0 as g],
aggr=[bit_or(distinct_simplify_t.v) as bit_or(DISTINCT distinct_simplify_t.v)]
+02)--RepartitionExec: partitioning=Hash([g@0], 4), input_partitions=1
+03)----AggregateExec: mode=Partial, gby=[g@0 as g],
aggr=[bit_or(distinct_simplify_t.v) as bit_or(DISTINCT distinct_simplify_t.v)]
+04)------DataSourceExec: partitions=1, partition_sizes=[1]
+
+# Negative case: bit_xor cancels duplicate pairs, so DISTINCT is kept and
+# SingleDistinctToGroupBy still rewrites the plan.
+query II
+SELECT g, bit_xor(DISTINCT v) FROM distinct_simplify_t GROUP BY g ORDER BY g;
+----
+1 6
+2 7
+
+query TT
+EXPLAIN SELECT g, bit_xor(DISTINCT v) FROM distinct_simplify_t GROUP BY g;
+----
+logical_plan
+01)Projection: distinct_simplify_t.g, bit_xor(alias1) AS bit_xor(DISTINCT
distinct_simplify_t.v)
+02)--Aggregate: groupBy=[[distinct_simplify_t.g]], aggr=[[bit_xor(alias1)]]
+03)----Aggregate: groupBy=[[distinct_simplify_t.g, distinct_simplify_t.v AS
alias1]], aggr=[[]]
+04)------TableScan: distinct_simplify_t projection=[g, v]
+physical_plan
+01)ProjectionExec: expr=[g@0 as g, bit_xor(alias1)@1 as bit_xor(DISTINCT
distinct_simplify_t.v)]
+02)--AggregateExec: mode=FinalPartitioned, gby=[g@0 as g],
aggr=[bit_xor(alias1)]
+03)----RepartitionExec: partitioning=Hash([g@0], 4), input_partitions=4
+04)------AggregateExec: mode=Partial, gby=[g@0 as g], aggr=[bit_xor(alias1)]
+05)--------AggregateExec: mode=FinalPartitioned, gby=[g@0 as g, alias1@1 as
alias1], aggr=[]
+06)----------RepartitionExec: partitioning=Hash([g@0, alias1@1], 4),
input_partitions=1
+07)------------AggregateExec: mode=Partial, gby=[g@0 as g, v@1 as alias1],
aggr=[]
+08)--------------DataSourceExec: partitions=1, partition_sizes=[1]
+
+# Negative case: count deduplicates for real
+query II
+SELECT g, count(DISTINCT v) FROM distinct_simplify_t GROUP BY g ORDER BY g;
+----
+1 2
+2 1
+
+query TT
+EXPLAIN SELECT g, count(DISTINCT v) FROM distinct_simplify_t GROUP BY g;
+----
+logical_plan
+01)Projection: distinct_simplify_t.g, count(alias1) AS count(DISTINCT
distinct_simplify_t.v)
+02)--Aggregate: groupBy=[[distinct_simplify_t.g]], aggr=[[count(alias1)]]
+03)----Aggregate: groupBy=[[distinct_simplify_t.g, distinct_simplify_t.v AS
alias1]], aggr=[[]]
+04)------TableScan: distinct_simplify_t projection=[g, v]
+physical_plan
+01)ProjectionExec: expr=[g@0 as g, count(alias1)@1 as count(DISTINCT
distinct_simplify_t.v)]
+02)--AggregateExec: mode=FinalPartitioned, gby=[g@0 as g], aggr=[count(alias1)]
+03)----RepartitionExec: partitioning=Hash([g@0], 4), input_partitions=4
+04)------AggregateExec: mode=Partial, gby=[g@0 as g], aggr=[count(alias1)]
+05)--------AggregateExec: mode=FinalPartitioned, gby=[g@0 as g, alias1@1 as
alias1], aggr=[]
+06)----------RepartitionExec: partitioning=Hash([g@0, alias1@1], 4),
input_partitions=1
+07)------------AggregateExec: mode=Partial, gby=[g@0 as g, v@1 as alias1],
aggr=[]
+08)--------------DataSourceExec: partitions=1, partition_sizes=[1]
+
+# Mixed: a node that still needs one DISTINCT keeps all of them, so this rule
+# cannot change which plans SingleDistinctToGroupBy rewrites
+query III
+SELECT g, min(DISTINCT v), sum(DISTINCT v) FROM distinct_simplify_t GROUP BY g
ORDER BY g;
+----
+1 3 8
+2 7 7
+
+query TT
+EXPLAIN SELECT g, min(DISTINCT v), sum(DISTINCT v) FROM distinct_simplify_t
GROUP BY g;
+----
+logical_plan
+01)Aggregate: groupBy=[[distinct_simplify_t.g]], aggr=[[min(DISTINCT
distinct_simplify_t.v), sum(DISTINCT CAST(distinct_simplify_t.v AS Int64))]]
+02)--TableScan: distinct_simplify_t projection=[g, v]
+physical_plan
+01)AggregateExec: mode=FinalPartitioned, gby=[g@0 as g], aggr=[min(DISTINCT
distinct_simplify_t.v), sum(DISTINCT distinct_simplify_t.v)]
+02)--RepartitionExec: partitioning=Hash([g@0], 4), input_partitions=1
+03)----AggregateExec: mode=Partial, gby=[g@0 as g], aggr=[min(DISTINCT
distinct_simplify_t.v), sum(DISTINCT distinct_simplify_t.v)]
+04)------DataSourceExec: partitions=1, partition_sizes=[1]
Review Comment:
This case does not test the gate. `v` is `INT`, so type coercion makes the
`sum` argument `CAST(v AS Int64)` while the `min` argument stays `v`. The two
aggregates then have different arguments, and `SingleDistinctToGroupBy` does
not rewrite the node on `main` either. The expected plan is the same on `main`
and on this branch. If the gate is removed later, this test still passes.
`count(DISTINCT v)` needs no cast. With it, `main` rewrites the node, and
this branch must keep that rewrite. The expected output below is from this
branch:
```suggestion
# Mixed: a node that still needs one DISTINCT keeps all of them, so this rule
# cannot change which plans SingleDistinctToGroupBy rewrites. `count` needs
the
# rewrite, and the plan below shows that it still happens.
query III
SELECT g, min(DISTINCT v), count(DISTINCT v) FROM distinct_simplify_t GROUP
BY g ORDER BY g;
----
1 3 2
2 7 1
query TT
EXPLAIN SELECT g, min(DISTINCT v), count(DISTINCT v) FROM
distinct_simplify_t GROUP BY g;
----
logical_plan
01)Projection: distinct_simplify_t.g, min(alias1) AS min(DISTINCT
distinct_simplify_t.v), count(alias1) AS count(DISTINCT distinct_simplify_t.v)
02)--Aggregate: groupBy=[[distinct_simplify_t.g]], aggr=[[min(alias1),
count(alias1)]]
03)----Aggregate: groupBy=[[distinct_simplify_t.g, distinct_simplify_t.v AS
alias1]], aggr=[[]]
04)------TableScan: distinct_simplify_t projection=[g, v]
physical_plan
01)ProjectionExec: expr=[g@0 as g, min(alias1)@1 as min(DISTINCT
distinct_simplify_t.v), count(alias1)@2 as count(DISTINCT
distinct_simplify_t.v)]
02)--AggregateExec: mode=FinalPartitioned, gby=[g@0 as g],
aggr=[min(alias1), count(alias1)]
03)----RepartitionExec: partitioning=Hash([g@0], 4), input_partitions=4
04)------AggregateExec: mode=Partial, gby=[g@0 as g], aggr=[min(alias1),
count(alias1)]
05)--------AggregateExec: mode=FinalPartitioned, gby=[g@0 as g, alias1@1 as
alias1], aggr=[]
06)----------RepartitionExec: partitioning=Hash([g@0, alias1@1], 4),
input_partitions=1
07)------------AggregateExec: mode=Partial, gby=[g@0 as g, v@1 as alias1],
aggr=[]
08)--------------DataSourceExec: partitions=1, partition_sizes=[1]
```
##########
datafusion/expr/src/udaf.rs:
##########
@@ -940,6 +945,20 @@ pub trait AggregateUDFImpl: Debug + DynEq + DynHash + Send
+ Sync + Any {
false
}
+ /// How this function treats the `DISTINCT` modifier.
+ ///
+ /// Return [`DistinctHandling::Ignored`] for duplicate-insensitive
+ /// functions so that `f(DISTINCT x)` is planned as `f(x)`.
+ ///
+ /// Return [`DistinctHandling::Unsupported`] if the accumulator neither
+ /// reads `is_distinct` nor is reached only after the planner has already
+ /// deduplicated the input. Nothing reads this variant yet: rejecting such
+ /// queries at planning time, rather than silently returning the
+ /// non-distinct answer, is a follow-up change.
Review Comment:
The criterion here depends on the optimizer. "Reached only after the planner
has already deduplicated the input" is true for `stddev` and false for
`approx_percentile_cont_with_weight`, although both reject `DISTINCT` with
`not_impl_err!`. The only difference is the number of arguments, which decides
whether `SingleDistinctToGroupBy` can rewrite the node. A UDAF author cannot
evaluate that, and the meaning of the tag changes when that rule improves.
Please make the variants describe the function only:
```suggestion
/// Return [`DistinctHandling::Unsupported`] if the accumulator does not
/// implement `DISTINCT`, that is, it does not read `is_distinct`, or it
/// rejects `DISTINCT` with an error. The planner then has to deduplicate
/// the input or reject the query. Nothing reads this variant yet:
/// rejecting such queries at planning time is a follow-up change.
```
##########
datafusion/optimizer/src/eliminate_aggregate_distinct.rs:
##########
@@ -0,0 +1,368 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! [`EliminateAggregateDistinct`] drops the `DISTINCT` modifier from aggregate
+//! functions that report [`DistinctHandling::Ignored`]
+
+use crate::optimizer::ApplyOrder;
+use crate::{OptimizerConfig, OptimizerRule};
+
+use datafusion_common::Result;
+use datafusion_common::tree_node::{Transformed, TreeNode, TreeNodeRecursion};
+use datafusion_expr::expr::AggregateFunction;
+use datafusion_expr::expr_rewriter::NamePreserver;
+use datafusion_expr::{DistinctHandling, Expr, LogicalPlan};
+
+/// Optimizer rule that removes a `DISTINCT` modifier that cannot change the
+/// result of the aggregate it is attached to.
+///
+/// `min`, `max`, `bool_and`, `bit_or` and friends have an idempotent merge, so
+/// `min(DISTINCT x)` and `min(x)` return the same value. Removing the flag
here
+/// keeps [`crate::single_distinct_to_groupby::SingleDistinctToGroupBy`] from
+/// rewriting the plan into an inner group by that only exists to deduplicate.
+///
+/// An aggregate states how it treats duplicates through
+/// [`datafusion_expr::AggregateUDFImpl::distinct_handling`].
+///
+/// ```text
+/// Aggregate: groupBy=[[g]], aggr=[[min(DISTINCT x)]]
+/// ```
+///
+/// becomes
+///
+/// ```text
+/// Aggregate: groupBy=[[g]], aggr=[[min(x) AS "min(DISTINCT x)"]]
+/// ```
+///
+/// The alias keeps the output schema unchanged so the parent projection still
+/// resolves.
+#[derive(Default, Debug)]
+pub struct EliminateAggregateDistinct {}
+
+impl EliminateAggregateDistinct {
+ pub fn new() -> Self {
+ Self {}
+ }
+}
+
+impl OptimizerRule for EliminateAggregateDistinct {
+ fn name(&self) -> &str {
+ "eliminate_aggregate_distinct"
+ }
+
+ fn apply_order(&self) -> Option<ApplyOrder> {
+ Some(ApplyOrder::BottomUp)
+ }
+
+ fn supports_rewrite(&self) -> bool {
+ true
+ }
+
+ fn rewrite(
+ &self,
+ plan: LogicalPlan,
+ _config: &dyn OptimizerConfig,
+ ) -> Result<Transformed<LogicalPlan>> {
+ // Aggregate expressions only appear on Aggregate nodes, so every other
+ // node is a cheap no-op. Window functions carry their own `distinct`
+ // flag and are out of scope.
+ let LogicalPlan::Aggregate(aggregate) = &plan else {
+ return Ok(Transformed::no(plan));
+ };
+ if !can_strip_every_distinct(&aggregate.aggr_expr)? {
+ return Ok(Transformed::no(plan));
+ }
+
+ // Dropping `DISTINCT` changes `Expr::schema_name`, and with it the
+ // output schema of the Aggregate, so restore the original name. The
+ // aggregate may sit under an alias that type coercion added, so walk
+ // the expression rather than matching only its root.
+ let name_preserver = NamePreserver::new(&plan);
+ plan.map_expressions(|expr| {
+ let saved_name = name_preserver.save(&expr);
+ expr.transform_down(strip_ignored_distinct)
+ .map(|t| t.update_data(|e| saved_name.restore(e)))
+ })
+ }
+}
+
+/// Whether the node has at least one `DISTINCT` and every one is `Ignored`.
+///
+/// Stripping only some of them would change which plans
+/// [`crate::single_distinct_to_groupby::SingleDistinctToGroupBy`] rewrites,
+/// since that rule keys off how many distinct aggregates a node has and
+/// whether they share one argument. This is conservative: `min(DISTINCT x),
+/// count(DISTINCT y)` keeps the `min` flag though no rewrite is possible.
Review Comment:
The reason given here is not the one that applies. After a partial strip,
the number of distinct aggregates and their arguments can still satisfy
`is_single_distinct_agg`. What stops the rewrite is the alias that this rule
adds: `is_single_distinct_agg` returns `false` for any `aggr_expr` element that
is not a bare `AggregateFunction`. Suggested wording:
```suggestion
/// If only some of them were stripped, an `Honored` `DISTINCT` could stay
/// beside a stripped aggregate. A stripped aggregate carries an alias, and
/// [`crate::single_distinct_to_groupby::SingleDistinctToGroupBy`] does not
/// rewrite a node whose `aggr_expr` contains an alias. So `min(DISTINCT x),
/// count(DISTINCT x)` would lose the rewrite that `count` needs. When every
/// `DISTINCT` is `Ignored`, none is left after stripping, so that rule has
/// nothing to rewrite. This is conservative: `min(DISTINCT x),
/// count(DISTINCT y)` keeps the `min` flag. That flag costs nothing at run
/// time, because `min` ignores `is_distinct` when it selects its
accumulator.
```
##########
datafusion/expr/src/udaf.rs:
##########
@@ -1713,6 +1736,31 @@ pub enum SetMonotonicity {
NotMonotonic,
}
+/// How an aggregate function treats the `DISTINCT` modifier.
+///
+/// Mathematically, `Ignored` means the function's merge operation is
+/// idempotent (its state forms a semilattice): f(S ⊎ S) = f(S), so
+/// removing duplicates from the input cannot change the result.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+#[non_exhaustive]
+pub enum DistinctHandling {
+ /// The result is the same with or without `DISTINCT`, so the planner
+ /// is free to drop it. `min`, `max`, `bool_and`, `bit_or`, ...
+ Ignored,
+ /// `DISTINCT` is applied, so the planner must leave it alone. Either the
+ /// accumulator reads `AccumulatorArgs::is_distinct` and deduplicates its
+ /// input (`count`, `sum`, `avg`, `var_samp`, `array_agg`, ...), or it
+ /// rejects `DISTINCT` and relies on `SingleDistinctToGroupBy` to
+ /// deduplicate the input first (`stddev`, `approx_median`, ...). The
latter
+ /// works only when that rewrite applies; otherwise the query errors. This
+ /// is the default.
+ Honored,
+ /// The accumulator does not implement `DISTINCT` and nothing deduplicates
+ /// the input for it, so `f(DISTINCT ...)` either errors or silently
+ /// returns the non-distinct answer. `corr`, `regr_*`, `nth_value`, ...
+ Unsupported,
Review Comment:
Same point as on the trait method: `Honored` should mean "the accumulator
deduplicates its own input", and nothing else. Whether a rewrite can rescue a
function that does not is a planner decision, and it belongs in the planner,
not in the tag. With this wording `stddev`, `stddev_pop` and `approx_median`
move to `Unsupported`, next to `approx_percentile_cont_with_weight`, which
already is. No behavior changes, because nothing reads `Unsupported` yet.
```suggestion
/// The accumulator reads `AccumulatorArgs::is_distinct` and deduplicates
/// its input, so the planner must leave the flag alone. `count`, `sum`,
/// `avg`, `var_samp`, `array_agg`, ... This is the default.
Honored,
/// The accumulator does not implement `DISTINCT`: it does not read
/// `is_distinct`, or it rejects `DISTINCT` with an error. The planner
has
/// to deduplicate the input first (today `SingleDistinctToGroupBy` does
/// that for single-argument functions) or reject the query. `stddev`,
/// `approx_median`, `corr`, `regr_*`, `nth_value`, ...
Unsupported,
```
##########
docs/source/library-user-guide/functions/adding-udfs.md:
##########
@@ -1116,6 +1116,25 @@ impl Accumulator for GeometricMean {
}
```
+### Declaring how an Aggregate UDF treats `DISTINCT`
+
+By default DataFusion assumes an aggregate honours the `DISTINCT` modifier,
which means the accumulator is expected to
Review Comment:
Nit: the rest of this file uses American spelling, and the variant is
spelled `Honored`.
```suggestion
By default DataFusion assumes an aggregate honors the `DISTINCT` modifier,
which means the accumulator is expected to
```
##########
datafusion/optimizer/src/eliminate_aggregate_distinct.rs:
##########
@@ -0,0 +1,368 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! [`EliminateAggregateDistinct`] drops the `DISTINCT` modifier from aggregate
+//! functions that report [`DistinctHandling::Ignored`]
+
+use crate::optimizer::ApplyOrder;
+use crate::{OptimizerConfig, OptimizerRule};
+
+use datafusion_common::Result;
+use datafusion_common::tree_node::{Transformed, TreeNode, TreeNodeRecursion};
+use datafusion_expr::expr::AggregateFunction;
+use datafusion_expr::expr_rewriter::NamePreserver;
+use datafusion_expr::{DistinctHandling, Expr, LogicalPlan};
+
+/// Optimizer rule that removes a `DISTINCT` modifier that cannot change the
+/// result of the aggregate it is attached to.
+///
+/// `min`, `max`, `bool_and`, `bit_or` and friends have an idempotent merge, so
+/// `min(DISTINCT x)` and `min(x)` return the same value. Removing the flag
here
+/// keeps [`crate::single_distinct_to_groupby::SingleDistinctToGroupBy`] from
+/// rewriting the plan into an inner group by that only exists to deduplicate.
+///
+/// An aggregate states how it treats duplicates through
+/// [`datafusion_expr::AggregateUDFImpl::distinct_handling`].
+///
+/// ```text
+/// Aggregate: groupBy=[[g]], aggr=[[min(DISTINCT x)]]
+/// ```
+///
+/// becomes
+///
+/// ```text
+/// Aggregate: groupBy=[[g]], aggr=[[min(x) AS "min(DISTINCT x)"]]
+/// ```
+///
+/// The alias keeps the output schema unchanged so the parent projection still
+/// resolves.
+#[derive(Default, Debug)]
+pub struct EliminateAggregateDistinct {}
+
+impl EliminateAggregateDistinct {
+ pub fn new() -> Self {
+ Self {}
+ }
+}
+
+impl OptimizerRule for EliminateAggregateDistinct {
+ fn name(&self) -> &str {
+ "eliminate_aggregate_distinct"
+ }
+
+ fn apply_order(&self) -> Option<ApplyOrder> {
+ Some(ApplyOrder::BottomUp)
+ }
+
+ fn supports_rewrite(&self) -> bool {
+ true
+ }
+
+ fn rewrite(
+ &self,
+ plan: LogicalPlan,
+ _config: &dyn OptimizerConfig,
+ ) -> Result<Transformed<LogicalPlan>> {
+ // Aggregate expressions only appear on Aggregate nodes, so every other
+ // node is a cheap no-op. Window functions carry their own `distinct`
+ // flag and are out of scope.
+ let LogicalPlan::Aggregate(aggregate) = &plan else {
+ return Ok(Transformed::no(plan));
+ };
+ if !can_strip_every_distinct(&aggregate.aggr_expr)? {
+ return Ok(Transformed::no(plan));
+ }
+
+ // Dropping `DISTINCT` changes `Expr::schema_name`, and with it the
+ // output schema of the Aggregate, so restore the original name. The
+ // aggregate may sit under an alias that type coercion added, so walk
+ // the expression rather than matching only its root.
+ let name_preserver = NamePreserver::new(&plan);
+ plan.map_expressions(|expr| {
+ let saved_name = name_preserver.save(&expr);
+ expr.transform_down(strip_ignored_distinct)
+ .map(|t| t.update_data(|e| saved_name.restore(e)))
+ })
+ }
+}
+
+/// Whether the node has at least one `DISTINCT` and every one is `Ignored`.
+///
+/// Stripping only some of them would change which plans
+/// [`crate::single_distinct_to_groupby::SingleDistinctToGroupBy`] rewrites,
+/// since that rule keys off how many distinct aggregates a node has and
+/// whether they share one argument. This is conservative: `min(DISTINCT x),
+/// count(DISTINCT y)` keeps the `min` flag though no rewrite is possible.
+fn can_strip_every_distinct(aggr_expr: &[Expr]) -> Result<bool> {
+ let mut handlings = vec![];
+ for expr in aggr_expr {
+ expr.apply(|e| {
+ if let Expr::AggregateFunction(AggregateFunction { func, params })
= e
+ && params.distinct
+ {
+ handlings.push(func.distinct_handling());
+ }
+ Ok(TreeNodeRecursion::Continue)
+ })?;
+ }
+ Ok(
+ !handlings.is_empty()
+ && handlings.iter().all(|h| *h == DistinctHandling::Ignored),
+ )
+}
Review Comment:
Nit: this runs on every `Aggregate` node of every query. It allocates a
`Vec` and walks all `aggr_expr` trees, and it does not stop at the first
`Honored` distinct. The note in `optimizer.rs` asks new rules for an aggressive
no-op path. One option:
```suggestion
fn can_strip_every_distinct(aggr_expr: &[Expr]) -> Result<bool> {
let mut found_distinct = false;
let mut all_ignored = true;
for expr in aggr_expr {
expr.apply(|e| {
if let Expr::AggregateFunction(AggregateFunction { func, params
}) = e
&& params.distinct
{
found_distinct = true;
if func.distinct_handling() != DistinctHandling::Ignored {
all_ignored = false;
return Ok(TreeNodeRecursion::Stop);
}
}
Ok(TreeNodeRecursion::Continue)
})?;
if !all_ignored {
break;
}
}
Ok(found_distinct && all_ignored)
}
```
##########
datafusion/functions-aggregate/src/bit_and_or_xor.rs:
##########
@@ -318,6 +319,18 @@ impl AggregateUDFImpl for BitwiseOperation {
fn documentation(&self) -> Option<&Documentation> {
Some(self.documentation)
}
+
+ fn distinct_handling(&self) -> DistinctHandling {
+ match self.operation {
+ // Bitwise AND/OR are idempotent: duplicates cannot change the
+ // result, so building a per-group `HashSet` buys nothing.
Review Comment:
Nit: AND and OR never build a `HashSet`. Only XOR has a distinct accumulator
(`DistinctBitXorAccumulator`), and `groups_accumulator_supported` already says
so.
```suggestion
// Bitwise AND/OR are idempotent: duplicates cannot change the
// result. Only XOR has a distinct accumulator.
```
##########
datafusion/functions-aggregate/src/first_last.rs:
##########
@@ -1294,6 +1300,12 @@ impl AggregateUDFImpl for LastValue {
) -> Result<Box<dyn GroupsAccumulator>> {
create_groups_accumulator(&args, false, self.is_input_pre_ordered,
self.name())
}
+
+ // TODO: whether this is `DistinctHandling::Ignored` depends on `ORDER BY`.
+ // `last_value(DISTINCT x ORDER BY y)` deduplicates `x` and leaves the `y`
+ // ordering meaningless, while `last_value(DISTINCT x ORDER BY x)` is just
+ // `max(x)`. Left at the default `Honored` until that is settled, even
Review Comment:
Nit: `last_value(DISTINCT x ORDER BY x)` is `max(x)` only when `x` has no
NULL. NULL sorts last, so `last_value` returns NULL where `max` skips it.
```suggestion
// ordering meaningless, while `last_value(DISTINCT x ORDER BY x)` is
// `max(x)` when `x` has no NULL. Left at the default `Honored` until
that is settled, even
```
##########
datafusion/functions-aggregate/src/stddev.rs:
##########
@@ -240,6 +246,12 @@ impl AggregateUDFImpl for StddevPop {
fn documentation(&self) -> Option<&Documentation> {
self.doc()
}
+
+ // Left at the default `Honored`. The accumulator rejects `DISTINCT` with
+ // `not_impl_err!`, so `f(DISTINCT x)` only works when
+ // `SingleDistinctToGroupBy` rewrites the node and deduplicates the input
+ // first. When it cannot, for example beside an `avg`, the query errors
+ // rather than returning the non-distinct answer.
Review Comment:
This accumulator rejects `DISTINCT` with `not_impl_err!`, the same as
`approx_percentile_cont_with_weight`, which is tagged `Unsupported`. With the
variant definitions suggested in `udaf.rs`, this one is `Unsupported` too. The
full path avoids a new import outside the diff; a `use` line is better.
```suggestion
fn distinct_handling(&self) -> datafusion_expr::DistinctHandling {
// The accumulator rejects `DISTINCT` with `not_impl_err!`, so the
// planner has to deduplicate the input first.
datafusion_expr::DistinctHandling::Unsupported
}
```
##########
datafusion/functions-aggregate/src/grouping.rs:
##########
@@ -110,4 +111,12 @@ impl AggregateUDFImpl for Grouping {
fn documentation(&self) -> Option<&Documentation> {
self.doc()
}
+
+ fn distinct_handling(&self) -> DistinctHandling {
+ // The result depends only on which grouping set a row belongs to, not
+ // on how many rows share a value, so duplicates cannot change it.
+ // `ResolveGroupingFunction` replaces the call before execution either
+ // way, which is why the accumulator above is never built.
Review Comment:
Nit: `ResolveGroupingFunction` is an analyzer rule. It replaces the call
before the optimizer runs, so this tag is not reachable from SQL. The tag is
still correct. Suggested wording:
```suggestion
// The result depends only on which grouping set a row belongs to,
not
// on how many rows share a value, so duplicates cannot change it.
// `ResolveGroupingFunction` replaces the call before the optimizer
// runs, so this tag is not reachable from SQL and the accumulator
// above is never built.
```
##########
datafusion/optimizer/src/eliminate_aggregate_distinct.rs:
##########
@@ -0,0 +1,368 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! [`EliminateAggregateDistinct`] drops the `DISTINCT` modifier from aggregate
+//! functions that report [`DistinctHandling::Ignored`]
+
+use crate::optimizer::ApplyOrder;
+use crate::{OptimizerConfig, OptimizerRule};
+
+use datafusion_common::Result;
+use datafusion_common::tree_node::{Transformed, TreeNode, TreeNodeRecursion};
+use datafusion_expr::expr::AggregateFunction;
+use datafusion_expr::expr_rewriter::NamePreserver;
+use datafusion_expr::{DistinctHandling, Expr, LogicalPlan};
+
+/// Optimizer rule that removes a `DISTINCT` modifier that cannot change the
+/// result of the aggregate it is attached to.
+///
+/// `min`, `max`, `bool_and`, `bit_or` and friends have an idempotent merge, so
+/// `min(DISTINCT x)` and `min(x)` return the same value. Removing the flag
here
+/// keeps [`crate::single_distinct_to_groupby::SingleDistinctToGroupBy`] from
+/// rewriting the plan into an inner group by that only exists to deduplicate.
+///
+/// An aggregate states how it treats duplicates through
+/// [`datafusion_expr::AggregateUDFImpl::distinct_handling`].
+///
+/// ```text
+/// Aggregate: groupBy=[[g]], aggr=[[min(DISTINCT x)]]
+/// ```
+///
+/// becomes
+///
+/// ```text
+/// Aggregate: groupBy=[[g]], aggr=[[min(x) AS "min(DISTINCT x)"]]
+/// ```
+///
+/// The alias keeps the output schema unchanged so the parent projection still
+/// resolves.
+#[derive(Default, Debug)]
+pub struct EliminateAggregateDistinct {}
+
+impl EliminateAggregateDistinct {
+ pub fn new() -> Self {
+ Self {}
+ }
+}
+
+impl OptimizerRule for EliminateAggregateDistinct {
+ fn name(&self) -> &str {
+ "eliminate_aggregate_distinct"
+ }
+
+ fn apply_order(&self) -> Option<ApplyOrder> {
+ Some(ApplyOrder::BottomUp)
+ }
+
+ fn supports_rewrite(&self) -> bool {
+ true
+ }
+
+ fn rewrite(
+ &self,
+ plan: LogicalPlan,
+ _config: &dyn OptimizerConfig,
+ ) -> Result<Transformed<LogicalPlan>> {
+ // Aggregate expressions only appear on Aggregate nodes, so every other
+ // node is a cheap no-op. Window functions carry their own `distinct`
+ // flag and are out of scope.
+ let LogicalPlan::Aggregate(aggregate) = &plan else {
+ return Ok(Transformed::no(plan));
+ };
+ if !can_strip_every_distinct(&aggregate.aggr_expr)? {
+ return Ok(Transformed::no(plan));
+ }
+
+ // Dropping `DISTINCT` changes `Expr::schema_name`, and with it the
+ // output schema of the Aggregate, so restore the original name. The
+ // aggregate may sit under an alias that type coercion added, so walk
+ // the expression rather than matching only its root.
+ let name_preserver = NamePreserver::new(&plan);
+ plan.map_expressions(|expr| {
+ let saved_name = name_preserver.save(&expr);
+ expr.transform_down(strip_ignored_distinct)
+ .map(|t| t.update_data(|e| saved_name.restore(e)))
+ })
+ }
+}
+
+/// Whether the node has at least one `DISTINCT` and every one is `Ignored`.
+///
+/// Stripping only some of them would change which plans
+/// [`crate::single_distinct_to_groupby::SingleDistinctToGroupBy`] rewrites,
+/// since that rule keys off how many distinct aggregates a node has and
+/// whether they share one argument. This is conservative: `min(DISTINCT x),
+/// count(DISTINCT y)` keeps the `min` flag though no rewrite is possible.
+fn can_strip_every_distinct(aggr_expr: &[Expr]) -> Result<bool> {
+ let mut handlings = vec![];
+ for expr in aggr_expr {
+ expr.apply(|e| {
+ if let Expr::AggregateFunction(AggregateFunction { func, params })
= e
+ && params.distinct
+ {
+ handlings.push(func.distinct_handling());
+ }
+ Ok(TreeNodeRecursion::Continue)
+ })?;
+ }
+ Ok(
+ !handlings.is_empty()
+ && handlings.iter().all(|h| *h == DistinctHandling::Ignored),
+ )
+}
+
+/// Drops `DISTINCT` from `expr` if it is an aggregate that ignores duplicates.
+///
+/// The handling is checked again here rather than trusted to
+/// [`can_strip_every_distinct`], which only inspects `aggr_expr`, while
+/// `map_expressions` also visits the group expressions.
+///
+/// An idempotent merge is also commutative, so an `Ignored` function is
+/// insensitive to input order and `order_by` needs no extra guard. `filter` is
+/// applied before deduplication either way, so it is carried over untouched.
Review Comment:
Idempotence does not imply commutativity. `first_value` is the
counterexample the PR itself uses. Also, `bit_and`, `bit_or` and
`approx_distinct` keep the default `HardRequirement` order sensitivity, so
"insensitive to input order" is not what they declare. The rule is still
correct, for a different reason. Suggested wording:
```suggestion
/// An idempotent merge is not always commutative: `first_value` is
/// idempotent but order-sensitive. The rule does not need commutativity.
/// `Ignored` means that the result does not change when duplicates are
/// removed, and stripping `DISTINCT` only stops that removal. `order_by` and
/// `filter` are carried over untouched, so the function sees the same rows
in
/// the same order, plus the duplicates that it ignores.
```
##########
datafusion/sqllogictest/test_files/aggregates_simplify.slt:
##########
@@ -356,3 +356,224 @@ DROP TABLE IF EXISTS tbl;
statement ok
DROP TABLE sum_simplify_t;
+
+#######
+# EliminateAggregateDistinct: DISTINCT is dropped from duplicate-insensitive
+# aggregates, so no inner group by is planned to deduplicate the input.
+#######
+
+statement ok
+CREATE TABLE distinct_simplify_t (g INT, v INT, b BOOLEAN) AS VALUES
+ (1, 3, true),
+ (1, 3, true),
+ (1, 5, false),
+ (2, 7, true),
+ (2, 7, true),
+ (2, NULL, NULL);
+
+# min: DISTINCT cannot change the minimum
+query II
+SELECT g, min(DISTINCT v) FROM distinct_simplify_t GROUP BY g ORDER BY g;
+----
+1 3
+2 7
+
+query TT
+EXPLAIN SELECT g, min(DISTINCT v) FROM distinct_simplify_t GROUP BY g;
+----
+logical_plan
+01)Aggregate: groupBy=[[distinct_simplify_t.g]],
aggr=[[min(distinct_simplify_t.v) AS min(DISTINCT distinct_simplify_t.v)]]
+02)--TableScan: distinct_simplify_t projection=[g, v]
+physical_plan
+01)AggregateExec: mode=FinalPartitioned, gby=[g@0 as g],
aggr=[min(distinct_simplify_t.v) as min(DISTINCT distinct_simplify_t.v)]
+02)--RepartitionExec: partitioning=Hash([g@0], 4), input_partitions=1
+03)----AggregateExec: mode=Partial, gby=[g@0 as g],
aggr=[min(distinct_simplify_t.v) as min(DISTINCT distinct_simplify_t.v)]
+04)------DataSourceExec: partitions=1, partition_sizes=[1]
+
+# max
+query II
+SELECT g, max(DISTINCT v) FROM distinct_simplify_t GROUP BY g ORDER BY g;
+----
+1 5
+2 7
+
+query TT
+EXPLAIN SELECT g, max(DISTINCT v) FROM distinct_simplify_t GROUP BY g;
+----
+logical_plan
+01)Aggregate: groupBy=[[distinct_simplify_t.g]],
aggr=[[max(distinct_simplify_t.v) AS max(DISTINCT distinct_simplify_t.v)]]
+02)--TableScan: distinct_simplify_t projection=[g, v]
+physical_plan
+01)AggregateExec: mode=FinalPartitioned, gby=[g@0 as g],
aggr=[max(distinct_simplify_t.v) as max(DISTINCT distinct_simplify_t.v)]
+02)--RepartitionExec: partitioning=Hash([g@0], 4), input_partitions=1
+03)----AggregateExec: mode=Partial, gby=[g@0 as g],
aggr=[max(distinct_simplify_t.v) as max(DISTINCT distinct_simplify_t.v)]
+04)------DataSourceExec: partitions=1, partition_sizes=[1]
+
+# bool_and
+query IB
+SELECT g, bool_and(DISTINCT b) FROM distinct_simplify_t GROUP BY g ORDER BY g;
+----
+1 false
+2 true
+
+query TT
+EXPLAIN SELECT g, bool_and(DISTINCT b) FROM distinct_simplify_t GROUP BY g;
+----
+logical_plan
+01)Aggregate: groupBy=[[distinct_simplify_t.g]],
aggr=[[bool_and(distinct_simplify_t.b) AS bool_and(DISTINCT
distinct_simplify_t.b)]]
+02)--TableScan: distinct_simplify_t projection=[g, b]
+physical_plan
+01)AggregateExec: mode=FinalPartitioned, gby=[g@0 as g],
aggr=[bool_and(distinct_simplify_t.b) as bool_and(DISTINCT
distinct_simplify_t.b)]
+02)--RepartitionExec: partitioning=Hash([g@0], 4), input_partitions=1
+03)----AggregateExec: mode=Partial, gby=[g@0 as g],
aggr=[bool_and(distinct_simplify_t.b) as bool_and(DISTINCT
distinct_simplify_t.b)]
+04)------DataSourceExec: partitions=1, partition_sizes=[1]
+
+# bool_or
+query IB
+SELECT g, bool_or(DISTINCT b) FROM distinct_simplify_t GROUP BY g ORDER BY g;
+----
+1 true
+2 true
+
+query TT
+EXPLAIN SELECT g, bool_or(DISTINCT b) FROM distinct_simplify_t GROUP BY g;
+----
+logical_plan
+01)Aggregate: groupBy=[[distinct_simplify_t.g]],
aggr=[[bool_or(distinct_simplify_t.b) AS bool_or(DISTINCT
distinct_simplify_t.b)]]
+02)--TableScan: distinct_simplify_t projection=[g, b]
+physical_plan
+01)AggregateExec: mode=FinalPartitioned, gby=[g@0 as g],
aggr=[bool_or(distinct_simplify_t.b) as bool_or(DISTINCT distinct_simplify_t.b)]
+02)--RepartitionExec: partitioning=Hash([g@0], 4), input_partitions=1
+03)----AggregateExec: mode=Partial, gby=[g@0 as g],
aggr=[bool_or(distinct_simplify_t.b) as bool_or(DISTINCT distinct_simplify_t.b)]
+04)------DataSourceExec: partitions=1, partition_sizes=[1]
+
+# bit_and
+query II
+SELECT g, bit_and(DISTINCT v) FROM distinct_simplify_t GROUP BY g ORDER BY g;
+----
+1 1
+2 7
+
+query TT
+EXPLAIN SELECT g, bit_and(DISTINCT v) FROM distinct_simplify_t GROUP BY g;
+----
+logical_plan
+01)Aggregate: groupBy=[[distinct_simplify_t.g]],
aggr=[[bit_and(distinct_simplify_t.v) AS bit_and(DISTINCT
distinct_simplify_t.v)]]
+02)--TableScan: distinct_simplify_t projection=[g, v]
+physical_plan
+01)AggregateExec: mode=FinalPartitioned, gby=[g@0 as g],
aggr=[bit_and(distinct_simplify_t.v) as bit_and(DISTINCT distinct_simplify_t.v)]
+02)--RepartitionExec: partitioning=Hash([g@0], 4), input_partitions=1
+03)----AggregateExec: mode=Partial, gby=[g@0 as g],
aggr=[bit_and(distinct_simplify_t.v) as bit_and(DISTINCT distinct_simplify_t.v)]
+04)------DataSourceExec: partitions=1, partition_sizes=[1]
+
+# bit_or
+query II
+SELECT g, bit_or(DISTINCT v) FROM distinct_simplify_t GROUP BY g ORDER BY g;
+----
+1 7
+2 7
+
+query TT
+EXPLAIN SELECT g, bit_or(DISTINCT v) FROM distinct_simplify_t GROUP BY g;
+----
+logical_plan
+01)Aggregate: groupBy=[[distinct_simplify_t.g]],
aggr=[[bit_or(distinct_simplify_t.v) AS bit_or(DISTINCT distinct_simplify_t.v)]]
+02)--TableScan: distinct_simplify_t projection=[g, v]
+physical_plan
+01)AggregateExec: mode=FinalPartitioned, gby=[g@0 as g],
aggr=[bit_or(distinct_simplify_t.v) as bit_or(DISTINCT distinct_simplify_t.v)]
+02)--RepartitionExec: partitioning=Hash([g@0], 4), input_partitions=1
+03)----AggregateExec: mode=Partial, gby=[g@0 as g],
aggr=[bit_or(distinct_simplify_t.v) as bit_or(DISTINCT distinct_simplify_t.v)]
+04)------DataSourceExec: partitions=1, partition_sizes=[1]
+
+# Negative case: bit_xor cancels duplicate pairs, so DISTINCT is kept and
+# SingleDistinctToGroupBy still rewrites the plan.
+query II
+SELECT g, bit_xor(DISTINCT v) FROM distinct_simplify_t GROUP BY g ORDER BY g;
+----
+1 6
+2 7
+
+query TT
+EXPLAIN SELECT g, bit_xor(DISTINCT v) FROM distinct_simplify_t GROUP BY g;
+----
+logical_plan
+01)Projection: distinct_simplify_t.g, bit_xor(alias1) AS bit_xor(DISTINCT
distinct_simplify_t.v)
+02)--Aggregate: groupBy=[[distinct_simplify_t.g]], aggr=[[bit_xor(alias1)]]
+03)----Aggregate: groupBy=[[distinct_simplify_t.g, distinct_simplify_t.v AS
alias1]], aggr=[[]]
+04)------TableScan: distinct_simplify_t projection=[g, v]
+physical_plan
+01)ProjectionExec: expr=[g@0 as g, bit_xor(alias1)@1 as bit_xor(DISTINCT
distinct_simplify_t.v)]
+02)--AggregateExec: mode=FinalPartitioned, gby=[g@0 as g],
aggr=[bit_xor(alias1)]
+03)----RepartitionExec: partitioning=Hash([g@0], 4), input_partitions=4
+04)------AggregateExec: mode=Partial, gby=[g@0 as g], aggr=[bit_xor(alias1)]
+05)--------AggregateExec: mode=FinalPartitioned, gby=[g@0 as g, alias1@1 as
alias1], aggr=[]
+06)----------RepartitionExec: partitioning=Hash([g@0, alias1@1], 4),
input_partitions=1
+07)------------AggregateExec: mode=Partial, gby=[g@0 as g, v@1 as alias1],
aggr=[]
+08)--------------DataSourceExec: partitions=1, partition_sizes=[1]
+
+# Negative case: count deduplicates for real
+query II
+SELECT g, count(DISTINCT v) FROM distinct_simplify_t GROUP BY g ORDER BY g;
+----
+1 2
+2 1
+
+query TT
+EXPLAIN SELECT g, count(DISTINCT v) FROM distinct_simplify_t GROUP BY g;
+----
+logical_plan
+01)Projection: distinct_simplify_t.g, count(alias1) AS count(DISTINCT
distinct_simplify_t.v)
+02)--Aggregate: groupBy=[[distinct_simplify_t.g]], aggr=[[count(alias1)]]
+03)----Aggregate: groupBy=[[distinct_simplify_t.g, distinct_simplify_t.v AS
alias1]], aggr=[[]]
+04)------TableScan: distinct_simplify_t projection=[g, v]
+physical_plan
+01)ProjectionExec: expr=[g@0 as g, count(alias1)@1 as count(DISTINCT
distinct_simplify_t.v)]
+02)--AggregateExec: mode=FinalPartitioned, gby=[g@0 as g], aggr=[count(alias1)]
+03)----RepartitionExec: partitioning=Hash([g@0], 4), input_partitions=4
+04)------AggregateExec: mode=Partial, gby=[g@0 as g], aggr=[count(alias1)]
+05)--------AggregateExec: mode=FinalPartitioned, gby=[g@0 as g, alias1@1 as
alias1], aggr=[]
+06)----------RepartitionExec: partitioning=Hash([g@0, alias1@1], 4),
input_partitions=1
+07)------------AggregateExec: mode=Partial, gby=[g@0 as g, v@1 as alias1],
aggr=[]
+08)--------------DataSourceExec: partitions=1, partition_sizes=[1]
+
+# Mixed: a node that still needs one DISTINCT keeps all of them, so this rule
+# cannot change which plans SingleDistinctToGroupBy rewrites
+query III
+SELECT g, min(DISTINCT v), sum(DISTINCT v) FROM distinct_simplify_t GROUP BY g
ORDER BY g;
+----
+1 3 8
+2 7 7
+
+query TT
+EXPLAIN SELECT g, min(DISTINCT v), sum(DISTINCT v) FROM distinct_simplify_t
GROUP BY g;
+----
+logical_plan
+01)Aggregate: groupBy=[[distinct_simplify_t.g]], aggr=[[min(DISTINCT
distinct_simplify_t.v), sum(DISTINCT CAST(distinct_simplify_t.v AS Int64))]]
+02)--TableScan: distinct_simplify_t projection=[g, v]
+physical_plan
+01)AggregateExec: mode=FinalPartitioned, gby=[g@0 as g], aggr=[min(DISTINCT
distinct_simplify_t.v), sum(DISTINCT distinct_simplify_t.v)]
+02)--RepartitionExec: partitioning=Hash([g@0], 4), input_partitions=1
+03)----AggregateExec: mode=Partial, gby=[g@0 as g], aggr=[min(DISTINCT
distinct_simplify_t.v), sum(DISTINCT distinct_simplify_t.v)]
+04)------DataSourceExec: partitions=1, partition_sizes=[1]
+
+# FILTER is applied before deduplication, so it survives the rewrite
+query II
+SELECT g, min(DISTINCT v) FILTER (WHERE v > 3) FROM distinct_simplify_t GROUP
BY g ORDER BY g;
+----
+1 5
+2 7
+
+query TT
+EXPLAIN SELECT g, min(DISTINCT v) FILTER (WHERE v > 3) FROM
distinct_simplify_t GROUP BY g;
+----
+logical_plan
+01)Aggregate: groupBy=[[distinct_simplify_t.g]],
aggr=[[min(distinct_simplify_t.v) FILTER (WHERE distinct_simplify_t.v >
Int32(3)) AS min(DISTINCT distinct_simplify_t.v) FILTER (WHERE
distinct_simplify_t.v > Int64(3))]]
+02)--TableScan: distinct_simplify_t projection=[g, v]
+physical_plan
+01)AggregateExec: mode=FinalPartitioned, gby=[g@0 as g],
aggr=[min(distinct_simplify_t.v) FILTER (WHERE distinct_simplify_t.v >
Int32(3)) as min(DISTINCT distinct_simplify_t.v) FILTER (WHERE
distinct_simplify_t.v > Int64(3))]
+02)--RepartitionExec: partitioning=Hash([g@0], 4), input_partitions=1
+03)----AggregateExec: mode=Partial, gby=[g@0 as g],
aggr=[min(distinct_simplify_t.v) FILTER (WHERE distinct_simplify_t.v >
Int32(3)) as min(DISTINCT distinct_simplify_t.v) FILTER (WHERE
distinct_simplify_t.v > Int64(3))]
+04)------DataSourceExec: partitions=1, partition_sizes=[1]
+
+statement ok
Review Comment:
This PR changes one visible result. On `main`, `max(DISTINCT v)` over
`(-1.0, -0.0)` prints `0.0`, because the inner group by turns `-0.0` into `0.0`
in its hash key. On this branch it prints `-0.0`, the same as `max(v)`. This is
an improvement, but no test pins it, and a test with an `R` column cannot catch
it because `R` prints both values as `0`. Please add a test that compares the
text form. Expected output below is from this branch. A sentence about this in
the PR description would also help.
```suggestion
# min/max(DISTINCT) over floats now sees the raw values. Before this rule,
the
# inner group by of SingleDistinctToGroupBy turned -0.0 into 0.0 in its hash
# key, so max(DISTINCT v) printed 0.0 here. Now it agrees with max(v).
# The R column type prints -0.0 as 0, so compare the text form.
statement ok
CREATE TABLE distinct_simplify_float_t (v DOUBLE) AS VALUES (-1.0), (-0.0),
(-0.0);
query TTTT
SELECT cast(min(DISTINCT v) AS VARCHAR), cast(max(DISTINCT v) AS VARCHAR),
cast(min(v) AS VARCHAR), cast(max(v) AS VARCHAR) FROM distinct_simplify_float_t;
----
-1.0 -0.0 -1.0 -0.0
statement ok
DROP TABLE distinct_simplify_float_t;
statement ok
```
##########
docs/source/library-user-guide/functions/adding-udfs.md:
##########
@@ -1116,6 +1116,25 @@ impl Accumulator for GeometricMean {
}
```
+### Declaring how an Aggregate UDF treats `DISTINCT`
+
+By default DataFusion assumes an aggregate honours the `DISTINCT` modifier,
which means the accumulator is expected to
+read `AccumulatorArgs::is_distinct` and deduplicate its input. Override
+[`AggregateUDFImpl::distinct_handling`] when that is not what your function
does:
+
+- Return `DistinctHandling::Ignored` when duplicates cannot change the result,
that is, when merging a value the
+ accumulator has already seen is a no-op. `min`, `max`, `bool_and` and
`bit_or` are all in this group. The optimizer
+ then plans `f(DISTINCT x)` as `f(x)`, which skips both the per-group hash
set and the extra grouping stage that
+ `SingleDistinctToGroupBy` would otherwise introduce.
+- Return `DistinctHandling::Unsupported` when the result depends on
duplicates, the accumulator does not deduplicate
+ its input, and nothing deduplicates it first, so `f(DISTINCT x)` errors or
silently returns the non-distinct answer.
+ Today this is a declaration only; rejecting such queries at planning time is
a follow-up change.
+- Leave the default `DistinctHandling::Honored` otherwise. That includes an
accumulator that rejects `DISTINCT` with an
+ error but relies on `SingleDistinctToGroupBy` to deduplicate the input
first, as `stddev` does: the query works when
+ that rewrite applies and errors when it does not.
Review Comment:
To match the variant definitions suggested in `udaf.rs`:
```suggestion
- Return `DistinctHandling::Unsupported` when the accumulator does not
implement `DISTINCT`: it does not read
`is_distinct`, or it rejects `DISTINCT` with an error. The planner must
then deduplicate the input first or reject
the query. Today this is a declaration only; rejecting such queries at
planning time is a follow-up change.
- Leave the default `DistinctHandling::Honored` when the accumulator reads
`AccumulatorArgs::is_distinct` and
deduplicates its input itself.
```
##########
datafusion/functions-aggregate/src/stddev.rs:
##########
@@ -140,6 +140,12 @@ impl AggregateUDFImpl for Stddev {
fn documentation(&self) -> Option<&Documentation> {
self.doc()
}
+
+ // Left at the default `Honored`. The accumulator rejects `DISTINCT` with
+ // `not_impl_err!`, so `f(DISTINCT x)` only works when
+ // `SingleDistinctToGroupBy` rewrites the node and deduplicates the input
+ // first. When it cannot, for example beside an `avg`, the query errors
+ // rather than returning the non-distinct answer.
Review Comment:
This accumulator rejects `DISTINCT` with `not_impl_err!`, the same as
`approx_percentile_cont_with_weight`, which is tagged `Unsupported`. With the
variant definitions suggested in `udaf.rs`, this one is `Unsupported` too. The
full path avoids a new import outside the diff; a `use` line is better.
```suggestion
fn distinct_handling(&self) -> datafusion_expr::DistinctHandling {
// The accumulator rejects `DISTINCT` with `not_impl_err!`, so the
// planner has to deduplicate the input first.
datafusion_expr::DistinctHandling::Unsupported
}
```
--
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]
