You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@datafusion.apache.org by gi...@apache.org on 2026/10/12 15:07:22 UTC
(datafusion) 01/01: fix: do not extract common subexpressions from the arguments of an aggregate with a FILTER (#26184)
This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] pushed a commit to branch gh-readonly-queue/main/pr-26184-a123121410afd050797000bfcd45fa57fdb80812
in repository https://gitbox.apache.org/repos/asf/datafusion.git
commit ee0b5c8b7217e85e6905f163edb0dee72f16ea92
Author: Adrian Garcia Badaracco <17...@users.noreply.github.com>
AuthorDate: Mon Oct 12 15:06:57 2026 +0000
fix: do not extract common subexpressions from the arguments of an aggregate with a FILTER (#26184)
## Which issue does this PR close?
- Closes #26183.
## Rationale for this change
An aggregate with a `FILTER` clause evaluates its arguments only for the
rows that pass the filter, so a fallible argument does not fail on a
rejected row.
When two aggregates have the same argument, `CommonSubexprEliminate`
moves that argument into a projection below the aggregate.
The projection evaluates it for every row, and the query fails on a row
that the filter rejects:
```sql
CREATE TABLE t AS
SELECT * FROM (VALUES (CAST(-950 AS SMALLINT), 0), (CAST(5 AS SMALLINT), 1)) AS v(x, y);
-- Succeeds.
SELECT avg(arrow_cast(x, 'UInt64')) FILTER (WHERE y > 0) AS a FROM t;
-- Fails: Arrow error: Cast error: Can't cast value -950 to type UInt64
SELECT
sum(arrow_cast(x, 'UInt64')) FILTER (WHERE y > 0) AS s,
count(arrow_cast(x, 'UInt64')) FILTER (WHERE y > 0) AS c
FROM t;
```
The rule already has a model for this: a conditionally evaluated
subexpression (a `CASE` branch, the right side of `AND` / `OR`, a lazy
function argument) is extracted only if it is also surely evaluated
somewhere else.
The arguments of a filtered aggregate are conditionally evaluated in the
same sense, but the rule did not know that.
## What changes are included in this PR?
1. `ExprCSEController::conditional_children` now reports the arguments
and the `ORDER BY` expressions of an `Expr::AggregateFunction` with a
`FILTER` as conditionally evaluated.
The filter itself stays a normal child, because the aggregate evaluates
it for every row.
A filter that several aggregates share is still extracted.
2. The same function reports the arguments of an `Expr::WindowFunction`
with a `FILTER` as conditionally evaluated.
The `PARTITION BY` expressions, the `ORDER BY` expressions and the
filter stay normal children.
The window executor does not yet evaluate the filter first (#24507,
#24508), so this part has no visible effect today.
It keeps the rule correct when that fix merges.
3. `CSEVisitor::f_down` in `datafusion/common/src/cse.rs` now visits the
children of a node with conditional children in tree order.
Before, it visited all normal children and then all conditional
children.
`CSERewriter` reads `id_array` in pre-order, so that was correct only
when every normal child came before every conditional child in the tree.
The children of an aggregate are `args`, `filter`, `order_by`, so the
normal child is in the middle.
The `CSEController` trait is unchanged.
The rule keeps every extraction that is safe:
| Query shape | Before | After |
|---|---|---|
| `sum(f(x)) FILTER (WHERE p), count(f(x)) FILTER (WHERE p)` | `f(x)`
and `p` extracted | only `p` extracted |
| `sum(f(x)) FILTER (WHERE p), count(f(x))` | `f(x)` extracted | `f(x)`
extracted (the second aggregate evaluates it for every row) |
| `sum(f(x)) FILTER (WHERE f(x) > 0)` | `f(x)` extracted | `f(x)`
extracted (the filter evaluates it for every row) |
| `sum(f(x)) FILTER (WHERE p) AS a, sum(f(x)) FILTER (WHERE p) AS b` |
one aggregate | one aggregate |
The cost is the first row: when an expression occurs only under filtered
aggregates, each aggregate now evaluates it, on the rows that pass its
filter.
An alternative keeps that extraction when every aggregate has the same
filter, with a guard such as `CASE WHEN p THEN f(x) END` in the
projection.
I did not do that here because it is a larger change with its own plan
shape, and it can be a follow-up.
## What is the testing strategy for this PR?
- Unit tests in `common_subexpr_eliminate.rs`:
`test_aggregate_filter_arguments` (the four rows of the table above),
`test_aggregate_filter_order_by` and `test_window_filter_arguments`.
- `cse.slt`: the reproduction above, with and without `GROUP BY`, the
`EXPLAIN` that shows the cast stays in the aggregate, and the case where
an unfiltered aggregate shares the argument.
- `aggregates_simplify.slt`: one expected plan changes. `SUM(column1 +
1) FILTER (WHERE column1 > 1), SUM(column1 + 1) FILTER (WHERE column1 >
0)` no longer extracts `column1 + 1`.
## Are there any user-facing changes?
Queries that failed with an error from a row that an aggregate `FILTER`
rejects now succeed.
Plans for queries where a subexpression occurs only in the arguments of
filtered aggregates no longer have that subexpression in a
`__common_expr` projection.
There are no API changes.
---
datafusion/common/src/cse.rs | 26 ++-
.../optimizer/src/common_subexpr_eliminate.rs | 231 ++++++++++++++++++++-
.../test_files/aggregates_simplify.slt | 8 +-
datafusion/sqllogictest/test_files/cse.slt | 72 +++++++
4 files changed, 318 insertions(+), 19 deletions(-)
diff --git a/datafusion/common/src/cse.rs b/datafusion/common/src/cse.rs
index 64dad4f5c0..7469563bd6 100644
--- a/datafusion/common/src/cse.rs
+++ b/datafusion/common/src/cse.rs
@@ -26,7 +26,7 @@ use crate::tree_node::{
TreeNodeVisitor,
};
use indexmap::IndexMap;
-use std::collections::HashMap;
+use std::collections::{HashMap, HashSet};
use std::hash::{BuildHasher, Hash, Hasher, RandomState};
use std::marker::PhantomData;
use std::sync::Arc;
@@ -176,6 +176,9 @@ pub trait CSEController {
/// Splits the children to normal and conditionally evaluated ones or returns `None`
/// if all are always evaluated.
+ ///
+ /// The returned references must point to the children of `node` themselves. The two
+ /// lists can be in any order.
fn conditional_children(node: &Self::Node) -> Option<ChildrenList<&Self::Node>>;
// Returns true if a node is valid. If a node is invalid then it can't be eliminated.
@@ -363,14 +366,21 @@ where
// If we are already in a node that can short-circuit then start new
// traversals on its normal conditional children.
match C::conditional_children(node) {
- Some((normal, conditional)) => {
- normal
- .into_iter()
- .try_for_each(|n| n.visit(self).map(|_| ()))?;
- self.conditional = true;
- conditional
+ Some((_normal, conditional)) => {
+ // Visit the children in tree order, whatever order the
+ // controller lists them in. `CSERewriter` reads `id_array` in
+ // pre-order, so the indexes that this visitor assigns must
+ // follow the same order.
+ let conditional = conditional
.into_iter()
- .try_for_each(|n| n.visit(self).map(|_| ()))?;
+ .map(std::ptr::from_ref)
+ .collect::<HashSet<_>>();
+ node.apply_children(|child| {
+ self.conditional =
+ conditional.contains(&std::ptr::from_ref(child));
+ child.visit(self)?;
+ Ok(TreeNodeRecursion::Continue)
+ })?;
self.conditional = false;
TreeNodeRecursion::Jump
diff --git a/datafusion/optimizer/src/common_subexpr_eliminate.rs b/datafusion/optimizer/src/common_subexpr_eliminate.rs
index 72615083d9..48501f4258 100644
--- a/datafusion/optimizer/src/common_subexpr_eliminate.rs
+++ b/datafusion/optimizer/src/common_subexpr_eliminate.rs
@@ -30,7 +30,10 @@ use datafusion_common::alias::AliasGenerator;
use datafusion_common::cse::{CSE, CSEController, FoundCommonNodes};
use datafusion_common::tree_node::{Transformed, TreeNode};
use datafusion_common::{Column, DFSchema, DFSchemaRef, Result, qualified_name};
-use datafusion_expr::expr::{Alias, HigherOrderFunction, ScalarFunction};
+use datafusion_expr::expr::{
+ AggregateFunction, AggregateFunctionParams, Alias, HigherOrderFunction,
+ ScalarFunction, WindowFunction, WindowFunctionParams,
+};
use datafusion_expr::logical_plan::{
Aggregate, Filter, LogicalPlan, Projection, Sort, Window,
};
@@ -691,6 +694,62 @@ impl CSEController for ExprCSEController<'_> {
.chain(else_expr.iter().map(|e| e.as_ref()))
.collect(),
)),
+
+ // An aggregate function with a `FILTER` clause evaluates the filter for
+ // every input row, but it evaluates its arguments only for the rows that
+ // pass the filter. Extracting an argument subexpression into a projection
+ // below the aggregate would evaluate it for the rejected rows too, which
+ // can raise an error (for example a failed cast or a division by zero)
+ // that the filter must prevent. The `ORDER BY` expressions of the
+ // aggregate are conservatively handled like the arguments.
+ Expr::AggregateFunction(AggregateFunction {
+ params:
+ AggregateFunctionParams {
+ args,
+ filter: Some(filter),
+ order_by,
+ ..
+ },
+ ..
+ }) => Some((
+ vec![filter.as_ref()],
+ args.iter()
+ .chain(order_by.iter().map(|sort| &sort.expr))
+ .collect(),
+ )),
+
+ // The arguments of a window function with a `FILTER` clause are
+ // conservatively handled like the arguments of a filtered aggregate.
+ // The window executor does not guarantee the same today:
+ // `AggregateWindowExpr::get_result_column` evaluates the arguments
+ // before it applies the filter. This rule must not rely on that order,
+ // so it does not extract a subexpression that occurs only in such
+ // arguments. The `PARTITION BY` and `ORDER BY` expressions of the
+ // window are evaluated for every input row.
+ Expr::WindowFunction(window_function) => {
+ let WindowFunction {
+ params:
+ WindowFunctionParams {
+ args,
+ partition_by,
+ order_by,
+ filter: Some(filter),
+ ..
+ },
+ ..
+ } = window_function.as_ref()
+ else {
+ return None;
+ };
+ Some((
+ partition_by
+ .iter()
+ .chain(order_by.iter().map(|sort| &sort.expr))
+ .chain([filter.as_ref()])
+ .collect(),
+ args.iter().collect(),
+ ))
+ }
_ => None,
}
}
@@ -842,9 +901,9 @@ mod test {
use arrow::datatypes::{DataType, Field, Schema};
use datafusion_expr::logical_plan::{JoinType, table_scan};
use datafusion_expr::{
- AccumulatorFactoryFunction, AggregateUDF, ColumnarValue, ScalarFunctionArgs,
- ScalarUDF, ScalarUDFImpl, Signature, SimpleAggregateUDF, Volatility,
- grouping_set, is_null, not,
+ AccumulatorFactoryFunction, AggregateUDF, ColumnarValue, ExprFunctionExt,
+ ScalarFunctionArgs, ScalarUDF, ScalarUDFImpl, Signature, SimpleAggregateUDF,
+ Volatility, WindowFunctionDefinition, grouping_set, is_null, not,
};
use datafusion_expr::{lit, logical_plan::builder::LogicalPlanBuilder};
@@ -853,7 +912,7 @@ mod test {
use crate::optimizer::OptimizerContext;
use crate::test::udfs::leaf_udf_expr;
use crate::test::*;
- use datafusion_expr::test::function_stub::{avg, sum};
+ use datafusion_expr::test::function_stub::{avg, count, sum, sum_udaf};
use datafusion_functions_window::row_number::row_number_udwf;
macro_rules! assert_optimized_plan_equal {
@@ -946,7 +1005,7 @@ mod test {
let return_type = DataType::UInt32;
let accumulator: AccumulatorFactoryFunction = Arc::new(|_| unimplemented!());
let udf_agg = |inner: Expr| {
- Expr::AggregateFunction(datafusion_expr::expr::AggregateFunction::new_udf(
+ Expr::AggregateFunction(AggregateFunction::new_udf(
Arc::new(AggregateUDF::from(SimpleAggregateUDF::new_with_signature(
"my_agg",
Signature::exact(vec![DataType::UInt32], Volatility::Stable),
@@ -1948,4 +2007,164 @@ mod test {
"
)
}
+
+ #[test]
+ fn test_aggregate_filter_arguments() -> Result<()> {
+ let table_scan = test_table_scan()?;
+
+ // An aggregate evaluates its arguments only for the rows that pass its
+ // `FILTER`. Thus `a + b` must stay in the aggregates. The filter `c > 0` is
+ // evaluated for every row and is extracted.
+ let filter = col("c").gt(lit(0));
+ let plan = LogicalPlanBuilder::from(table_scan.clone())
+ .aggregate(
+ iter::empty::<Expr>(),
+ vec![
+ sum(col("a") + col("b")).filter(filter.clone()).build()?,
+ count(col("a") + col("b")).filter(filter.clone()).build()?,
+ ],
+ )?
+ .build()?;
+
+ assert_optimized_plan_equal!(
+ plan,
+ @ "
+ Aggregate: groupBy=[[]], aggr=[[sum(test.a + test.b) FILTER (WHERE __common_expr_1 AS test.c > Int32(0)) AS sum(test.a + test.b) FILTER (WHERE test.c > Int32(0)), COUNT(test.a + test.b) FILTER (WHERE __common_expr_1 AS test.c > Int32(0)) AS COUNT(test.a + test.b) FILTER (WHERE test.c > Int32(0))]]
+ Projection: test.c > Int32(0) AS __common_expr_1, test.a, test.b, test.c
+ TableScan: test
+ "
+ )?;
+
+ // `a + b` is surely evaluated by the aggregate without a `FILTER`, so it is
+ // extracted.
+ let plan = LogicalPlanBuilder::from(table_scan.clone())
+ .aggregate(
+ iter::empty::<Expr>(),
+ vec![
+ sum(col("a") + col("b")).filter(filter.clone()).build()?,
+ count(col("a") + col("b")),
+ ],
+ )?
+ .build()?;
+
+ assert_optimized_plan_equal!(
+ plan,
+ @ "
+ Aggregate: groupBy=[[]], aggr=[[sum(__common_expr_1 AS test.a + test.b) FILTER (WHERE test.c > Int32(0)), COUNT(__common_expr_1 AS test.a + test.b)]]
+ Projection: test.a + test.b AS __common_expr_1, test.a, test.b, test.c
+ TableScan: test
+ "
+ )?;
+
+ // `a + b` is surely evaluated by the filter, so it is extracted.
+ let plan = LogicalPlanBuilder::from(table_scan.clone())
+ .aggregate(
+ iter::empty::<Expr>(),
+ vec![
+ sum(col("a") + col("b"))
+ .filter((col("a") + col("b")).gt(lit(0)))
+ .build()?,
+ ],
+ )?
+ .build()?;
+
+ assert_optimized_plan_equal!(
+ plan,
+ @ "
+ Aggregate: groupBy=[[]], aggr=[[sum(__common_expr_1 AS test.a + test.b) FILTER (WHERE __common_expr_1 AS test.a + test.b > Int32(0)) AS sum(test.a + test.b) FILTER (WHERE test.a + test.b > Int32(0))]]
+ Projection: test.a + test.b AS __common_expr_1, test.a, test.b, test.c
+ TableScan: test
+ "
+ )?;
+
+ // Identical aggregates with a `FILTER` are still deduplicated.
+ let plan = LogicalPlanBuilder::from(table_scan)
+ .aggregate(
+ iter::empty::<Expr>(),
+ vec![
+ sum(col("a") + col("b"))
+ .filter(filter.clone())
+ .build()?
+ .alias("s1"),
+ sum(col("a") + col("b")).filter(filter).build()?.alias("s2"),
+ ],
+ )?
+ .build()?;
+
+ assert_optimized_plan_equal!(
+ plan,
+ @ "
+ Projection: __common_expr_2 AS s1, __common_expr_2 AS s2
+ Aggregate: groupBy=[[]], aggr=[[sum(test.a + test.b) FILTER (WHERE __common_expr_1) AS __common_expr_2]]
+ Projection: test.c > Int32(0) AS __common_expr_1, test.a, test.b, test.c
+ TableScan: test
+ "
+ )
+ }
+
+ #[test]
+ fn test_aggregate_filter_order_by() -> Result<()> {
+ let table_scan = test_table_scan()?;
+
+ // The `ORDER BY` expressions of an aggregate with a `FILTER` are handled
+ // like its arguments, so `a + b` is not extracted.
+ let filter = col("c").gt(lit(0));
+ let order_by = vec![(col("a") + col("b")).sort(true, false)];
+ let plan = LogicalPlanBuilder::from(table_scan)
+ .aggregate(
+ iter::empty::<Expr>(),
+ vec![
+ sum(col("a"))
+ .order_by(order_by.clone())
+ .filter(filter.clone())
+ .build()?,
+ sum(col("b")).order_by(order_by).filter(filter).build()?,
+ ],
+ )?
+ .build()?;
+
+ assert_optimized_plan_equal!(
+ plan,
+ @ "
+ Aggregate: groupBy=[[]], aggr=[[sum(test.a) FILTER (WHERE __common_expr_1 AS test.c > Int32(0)) ORDER BY [test.a + test.b ASC NULLS LAST] AS sum(test.a) FILTER (WHERE test.c > Int32(0)) ORDER BY [test.a + test.b ASC NULLS LAST], sum(test.b) FILTER (WHERE __common_expr_1 AS test.c > Int32(0)) ORDER BY [test.a + test.b ASC NULLS LAST] AS sum(test.b) FILTER (WHERE test.c > Int32(0)) ORDER BY [test.a + test.b ASC NULLS LAST]]]
+ Projection: test.c > Int32(0) AS __common_expr_1, test.a, test.b, test.c
+ TableScan: test
+ "
+ )
+ }
+
+ #[test]
+ fn test_window_filter_arguments() -> Result<()> {
+ let table_scan = test_table_scan()?;
+
+ // CSE conservatively treats the arguments of a window function with a
+ // `FILTER` as conditionally evaluated, although the window executor
+ // evaluates them before it applies the filter today. The `PARTITION BY`
+ // expressions and the filter are evaluated for every input row.
+ let window_sum = |arg: Expr| {
+ Expr::from(WindowFunction::new(
+ WindowFunctionDefinition::AggregateUDF(sum_udaf()),
+ vec![arg],
+ ))
+ .partition_by(vec![col("a") * col("b")])
+ .filter(col("c").gt(lit(0)))
+ .build()
+ };
+ let plan = LogicalPlanBuilder::from(table_scan)
+ .window(vec![
+ window_sum(col("a") + col("b"))?,
+ window_sum((col("a") + col("b")) * lit(2))?,
+ ])?
+ .build()?;
+
+ assert_optimized_plan_equal!(
+ plan,
+ @ "
+ Projection: test.a, test.b, test.c, sum(test.a + test.b) FILTER (WHERE test.c > Int32(0)) PARTITION BY [test.a * test.b] ROWS BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING, sum(test.a + test.b * Int32(2)) FILTER (WHERE test.c > Int32(0)) PARTITION BY [test.a * test.b] ROWS BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING
+ WindowAggr: windowExpr=[[sum(test.a + test.b) FILTER (WHERE __common_expr_2 AS test.c > Int32(0)) PARTITION BY [__common_expr_1 AS test.a * test.b] ROWS BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING AS sum(test.a + test.b) FILTER (WHERE test.c > Int32(0)) PARTITION BY [test.a * test.b] ROWS BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING, sum((test.a + test.b) * Int32(2)) FILTER (WHERE __common_expr_2 AS test.c > Int32(0)) PARTITION BY [__common_expr_1 AS test.a * tes [...]
+ Projection: test.a * test.b AS __common_expr_1, test.c > Int32(0) AS __common_expr_2, test.a, test.b, test.c
+ TableScan: test
+ "
+ )
+ }
}
diff --git a/datafusion/sqllogictest/test_files/aggregates_simplify.slt b/datafusion/sqllogictest/test_files/aggregates_simplify.slt
index 4169707913..e5a50f30f3 100644
--- a/datafusion/sqllogictest/test_files/aggregates_simplify.slt
+++ b/datafusion/sqllogictest/test_files/aggregates_simplify.slt
@@ -226,13 +226,11 @@ query TT
EXPLAIN SELECT SUM(column1 + 1) FILTER (WHERE column1 > 1), SUM(column1 + 1) FILTER (WHERE column1 > 0) FROM sum_simplify_t;
----
logical_plan
-01)Aggregate: groupBy=[[]], aggr=[[sum(__common_expr_1 AS sum_simplify_t.column1 + Int64(1)) FILTER (WHERE sum_simplify_t.column1 > Int64(1)), sum(__common_expr_1 AS sum_simplify_t.column1 + Int64(1)) FILTER (WHERE sum_simplify_t.column1 > Int64(0))]]
-02)--Projection: sum_simplify_t.column1 + Int64(1) AS __common_expr_1, sum_simplify_t.column1
-03)----TableScan: sum_simplify_t projection=[column1]
+01)Aggregate: groupBy=[[]], aggr=[[sum(sum_simplify_t.column1 + Int64(1)) FILTER (WHERE sum_simplify_t.column1 > Int64(1)), sum(sum_simplify_t.column1 + Int64(1)) FILTER (WHERE sum_simplify_t.column1 > Int64(0))]]
+02)--TableScan: sum_simplify_t projection=[column1]
physical_plan
01)AggregateExec: mode=Single, gby=[], aggr=[sum(sum_simplify_t.column1 + Int64(1)) FILTER (WHERE sum_simplify_t.column1 > Int64(1)), sum(sum_simplify_t.column1 + Int64(1)) FILTER (WHERE sum_simplify_t.column1 > Int64(0))]
-02)--ProjectionExec: expr=[column1@0 + 1 as __common_expr_1, column1@0 as column1]
-03)----DataSourceExec: partitions=1, partition_sizes=[1]
+02)--DataSourceExec: partitions=1, partition_sizes=[1]
# volatile aggregate arguments
query B
diff --git a/datafusion/sqllogictest/test_files/cse.slt b/datafusion/sqllogictest/test_files/cse.slt
index 035c89079a..6f8abd8085 100644
--- a/datafusion/sqllogictest/test_files/cse.slt
+++ b/datafusion/sqllogictest/test_files/cse.slt
@@ -231,3 +231,75 @@ logical_plan
physical_plan
01)ProjectionExec: expr=[(random() = 0 OR a@0 = 1) AND a@0 = 2 as c1, random() = 0 AND a@0 = 2 OR a@0 = 1 as c2, CASE WHEN random() = 0 THEN a@0 + 3 ELSE a@0 + 3 END as c3, CASE WHEN random() = 0 THEN 0 WHEN a@0 + 4 = 0 THEN a@0 + 4 ELSE 0 END as c4, CASE WHEN random() = 0 THEN 0 WHEN a@0 + 5 = 0 THEN 0 ELSE a@0 + 5 END as c5, CASE WHEN random() = 0 THEN 0 WHEN random() = 0 THEN a@0 + 6 ELSE a@0 + 6 END as c6]
02)--DataSourceExec: partitions=1, partition_sizes=[0]
+
+# The arguments of an aggregate with a FILTER are evaluated only for the rows
+# that pass the filter. They must not be extracted into a projection, which
+# evaluates them for every row.
+statement ok
+CREATE TABLE cse_agg_filter AS
+SELECT * FROM (VALUES (CAST(-950 AS SMALLINT), 0), (CAST(5 AS SMALLINT), 1)) AS v(x, y)
+
+query TT
+EXPLAIN SELECT
+ sum(arrow_cast(x, 'UInt64')) FILTER (WHERE y > 0) AS s,
+ count(arrow_cast(x, 'UInt64')) FILTER (WHERE y > 0) AS c
+FROM cse_agg_filter
+----
+logical_plan
+01)Projection: sum(arrow_cast(cse_agg_filter.x,Utf8("UInt64"))) FILTER (WHERE cse_agg_filter.y > Int64(0)) AS s, count(arrow_cast(cse_agg_filter.x,Utf8("UInt64"))) FILTER (WHERE cse_agg_filter.y > Int64(0)) AS c
+02)--Aggregate: groupBy=[[]], aggr=[[sum(CAST(cse_agg_filter.x AS UInt64)) FILTER (WHERE __common_expr_1) AS sum(arrow_cast(cse_agg_filter.x,Utf8("UInt64"))) FILTER (WHERE cse_agg_filter.y > Int64(0)), count(CAST(cse_agg_filter.x AS UInt64)) FILTER (WHERE __common_expr_1) AS count(arrow_cast(cse_agg_filter.x,Utf8("UInt64"))) FILTER (WHERE cse_agg_filter.y > Int64(0))]]
+03)----Projection: cse_agg_filter.y > Int64(0) AS __common_expr_1, cse_agg_filter.x
+04)------TableScan: cse_agg_filter projection=[x, y]
+physical_plan
+01)ProjectionExec: expr=[sum(arrow_cast(cse_agg_filter.x,Utf8("UInt64"))) FILTER (WHERE cse_agg_filter.y > Int64(0))@0 as s, count(arrow_cast(cse_agg_filter.x,Utf8("UInt64"))) FILTER (WHERE cse_agg_filter.y > Int64(0))@1 as c]
+02)--AggregateExec: mode=Single, gby=[], aggr=[sum(cse_agg_filter.x) FILTER (WHERE __common_expr_1) as sum(arrow_cast(cse_agg_filter.x,Utf8("UInt64"))) FILTER (WHERE cse_agg_filter.y > Int64(0)), count(cse_agg_filter.x) FILTER (WHERE __common_expr_1) as count(arrow_cast(cse_agg_filter.x,Utf8("UInt64"))) FILTER (WHERE cse_agg_filter.y > Int64(0))]
+03)----ProjectionExec: expr=[y@1 > 0 as __common_expr_1, x@0 as x]
+04)------DataSourceExec: partitions=1, partition_sizes=[1]
+
+query II
+SELECT
+ sum(arrow_cast(x, 'UInt64')) FILTER (WHERE y > 0) AS s,
+ count(arrow_cast(x, 'UInt64')) FILTER (WHERE y > 0) AS c
+FROM cse_agg_filter
+----
+5 1
+
+query III
+SELECT
+ y,
+ sum(arrow_cast(x, 'UInt64')) FILTER (WHERE y > 0) AS s,
+ count(arrow_cast(x, 'UInt64')) FILTER (WHERE y > 0) AS c
+FROM cse_agg_filter
+GROUP BY y
+ORDER BY y
+----
+0 NULL 0
+1 5 1
+
+# An aggregate without a FILTER evaluates the common argument for every row,
+# so the argument is extracted and the cast error is the correct result.
+query TT
+EXPLAIN SELECT
+ sum(arrow_cast(x, 'UInt64')) FILTER (WHERE y > 0) AS s,
+ count(arrow_cast(x, 'UInt64')) AS c
+FROM cse_agg_filter
+----
+logical_plan
+01)Projection: sum(arrow_cast(cse_agg_filter.x,Utf8("UInt64"))) FILTER (WHERE cse_agg_filter.y > Int64(0)) AS s, count(arrow_cast(cse_agg_filter.x,Utf8("UInt64"))) AS c
+02)--Aggregate: groupBy=[[]], aggr=[[sum(__common_expr_1) FILTER (WHERE cse_agg_filter.y > Int64(0)) AS sum(arrow_cast(cse_agg_filter.x,Utf8("UInt64"))) FILTER (WHERE cse_agg_filter.y > Int64(0)), count(__common_expr_1) AS count(arrow_cast(cse_agg_filter.x,Utf8("UInt64")))]]
+03)----Projection: CAST(cse_agg_filter.x AS UInt64) AS __common_expr_1, cse_agg_filter.y
+04)------TableScan: cse_agg_filter projection=[x, y]
+physical_plan
+01)ProjectionExec: expr=[sum(arrow_cast(cse_agg_filter.x,Utf8("UInt64"))) FILTER (WHERE cse_agg_filter.y > Int64(0))@0 as s, count(arrow_cast(cse_agg_filter.x,Utf8("UInt64")))@1 as c]
+02)--AggregateExec: mode=Single, gby=[], aggr=[sum(__common_expr_1) FILTER (WHERE cse_agg_filter.y > Int64(0)) as sum(arrow_cast(cse_agg_filter.x,Utf8("UInt64"))) FILTER (WHERE cse_agg_filter.y > Int64(0)), count(__common_expr_1) as count(arrow_cast(cse_agg_filter.x,Utf8("UInt64")))]
+03)----ProjectionExec: expr=[CAST(x@0 AS UInt64) as __common_expr_1, y@1 as y]
+04)------DataSourceExec: partitions=1, partition_sizes=[1]
+
+query error Can't cast value \-950 to type UInt64
+SELECT
+ sum(arrow_cast(x, 'UInt64')) FILTER (WHERE y > 0) AS s,
+ count(arrow_cast(x, 'UInt64')) AS c
+FROM cse_agg_filter
+
+statement ok
+DROP TABLE cse_agg_filter
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@datafusion.apache.org
For additional commands, e-mail: commits-help@datafusion.apache.org