jayzhan211 commented on code in PR #25385:
URL: https://github.com/apache/datafusion/pull/25385#discussion_r4036753932
##########
datafusion/optimizer/src/utils.rs:
##########
@@ -39,6 +41,41 @@ use std::sync::Arc;
/// as it was initially placed here and then moved elsewhere.
pub use datafusion_expr::expr_rewriter::NamePreserver;
+/// Whether an expression is free of volatile scalar functions and subqueries.
+/// Subqueries are conservative barriers because their plans may contain
+/// volatile expressions that [`Expr::is_volatile`] does not visit.
+pub(crate) fn is_repeatable(expr: &Expr) -> bool {
+ !expr
+ .exists(|expr| {
+ Ok(expr.is_volatile_node()
+ || matches!(
+ expr,
+ Expr::Exists(_)
+ | Expr::InSubquery(_)
+ | Expr::SetComparison(_)
+ | Expr::ScalarSubquery(_)
+ ))
+ })
+ .expect("expression traversal is infallible")
+}
+
+/// Whether an aggregate explicitly declares duplicate insensitivity and can
+/// safely ignore repeated input rows. Arguments, FILTER, and ORDER BY must
+/// also be repeatable: MIN(random()) still observes repetitions.
+pub(crate) fn is_duplicate_insensitive_aggregate(mut expr: &Expr) -> bool {
+ while let Expr::Alias(alias) = expr {
+ expr = &alias.expr;
+ }
+ let Expr::AggregateFunction(aggregate) = expr else {
+ return false;
+ };
+ // Expr::is_volatile checks scalar functions only; check the aggregate
+ // function's own volatility separately.
+ aggregate.func.distinct_handling() == DistinctHandling::Insensitive
Review Comment:
```suggestion
(aggregate.params.distinct || aggregate.func.distinct_handling() ==
DistinctHandling::Insensitive)
```
##########
datafusion/optimizer/src/utils.rs:
##########
@@ -39,6 +41,41 @@ use std::sync::Arc;
/// as it was initially placed here and then moved elsewhere.
pub use datafusion_expr::expr_rewriter::NamePreserver;
+/// Whether an expression is free of volatile scalar functions and subqueries.
+/// Subqueries are conservative barriers because their plans may contain
+/// volatile expressions that [`Expr::is_volatile`] does not visit.
+pub(crate) fn is_repeatable(expr: &Expr) -> bool {
+ !expr
+ .exists(|expr| {
+ Ok(expr.is_volatile_node()
+ || matches!(
+ expr,
+ Expr::Exists(_)
+ | Expr::InSubquery(_)
+ | Expr::SetComparison(_)
+ | Expr::ScalarSubquery(_)
+ ))
+ })
+ .expect("expression traversal is infallible")
+}
+
+/// Whether an aggregate explicitly declares duplicate insensitivity and can
+/// safely ignore repeated input rows. Arguments, FILTER, and ORDER BY must
+/// also be repeatable: MIN(random()) still observes repetitions.
+pub(crate) fn is_duplicate_insensitive_aggregate(mut expr: &Expr) -> bool {
+ while let Expr::Alias(alias) = expr {
+ expr = &alias.expr;
+ }
+ let Expr::AggregateFunction(aggregate) = expr else {
+ return false;
+ };
+ // Expr::is_volatile checks scalar functions only; check the aggregate
+ // function's own volatility separately.
+ aggregate.func.distinct_handling() == DistinctHandling::Insensitive
Review Comment:
Should we check `distinct`?
--
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]
