You are viewing a plain text version of this content. The canonical link for it is here.
Posted to github@datafusion.apache.org by "UBarney (via GitHub)" <gi...@apache.org> on 2025/05/06 07:10:26 UTC
[PR] implement `AggregateExec.partition_statistics` [datafusion]
UBarney opened a new pull request, #15954:
URL: https://github.com/apache/datafusion/pull/15954
## Which issue does this PR close?
part of #15873
<!--
We generally require a GitHub issue to be filed for all bug fixes and enhancements and this helps us generate change logs for our releases. You can link an issue to this PR using the GitHub syntax. For example `Closes #123` indicates that this PR will close issue #123.
-->
- Closes #.
## Rationale for this change
<!--
Why are you proposing this change? If this is already explained clearly in the issue then this section is not needed.
Explaining clearly why changes are proposed helps reviewers understand your changes and offer better suggestions for fixes.
-->
## What changes are included in this PR?
- `statistics_inner`:
- Change signature to `(self, statistics: Statistics)`.
- Preserve `min_value`/`max_value` from input statistics when grouping by a simple `Column`.
- Add `PartitionStatisticsExec`
<!--
There is no need to duplicate the description in the issue here but it is sometimes worth providing a summary of the individual changes in this PR.
-->
## Are these changes tested?
Yes
<!--
We typically require tests for all PRs in order to:
1. Prevent the code from being accidentally broken by subsequent changes
2. Serve as another way to document the expected behavior of the code
If tests are not included in your PR, please explain why (for example, are they covered by existing tests)?
-->
## Are there any user-facing changes?
<!--
If there are user-facing changes then we may require documentation to be updated before approving the PR.
-->
<!--
If there are any breaking changes to public APIs, please add the `api change` label.
-->
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] implement `AggregateExec.partition_statistics` [datafusion]
Posted by "xudong963 (via GitHub)" <gi...@apache.org>.
xudong963 commented on code in PR #15954:
URL: https://github.com/apache/datafusion/pull/15954#discussion_r2081994480
##########
datafusion/physical-plan/src/test/exec.rs:
##########
@@ -614,6 +614,103 @@ impl ExecutionPlan for StatisticsExec {
}
}
+/// A mock execution plan that returns per-partition statistics
Review Comment:
Maybe you can just create `DataSourceExec` with statistics to do tests. https://github.com/apache/datafusion/blob/main/datafusion/core/tests/physical_optimizer/partition_statistics.rs#L189
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] implement `AggregateExec.partition_statistics` [datafusion]
Posted by "xudong963 (via GitHub)" <gi...@apache.org>.
xudong963 commented on code in PR #15954:
URL: https://github.com/apache/datafusion/pull/15954#discussion_r2102238189
##########
datafusion/physical-plan/src/aggregates/mod.rs:
##########
@@ -751,28 +771,16 @@ impl AggregateExec {
})
}
_ => {
- // When the input row count is 0 or 1, we can adopt that statistic keeping its reliability.
+ // When the input row count is 1, we can adopt that statistic keeping its reliability.
// When it is larger than 1, we degrade the precision since it may decrease after aggregation.
- let num_rows = if let Some(value) = self
- .input()
- .partition_statistics(None)?
- .num_rows
- .get_value()
+ let num_rows = if let Some(value) = child_statistics.num_rows.get_value()
{
- if *value > 1 {
- self.input()
- .partition_statistics(None)?
- .num_rows
- .to_inexact()
- } else if *value == 0 {
- // Aggregation on an empty table creates a null row.
Review Comment:
Does it seem like a bug fix? If so, it's better add a unit test to cover it
##########
datafusion/physical-plan/src/aggregates/mod.rs:
##########
@@ -751,28 +771,18 @@ impl AggregateExec {
})
}
_ => {
- // When the input row count is 0 or 1, we can adopt that statistic keeping its reliability.
+ // When the input row count is 1, we can adopt that statistic keeping its reliability.
// When it is larger than 1, we degrade the precision since it may decrease after aggregation.
- let num_rows = if let Some(value) = self
- .input()
- .partition_statistics(None)?
- .num_rows
- .get_value()
+ let num_rows = if let Some(value) = child_statistics.num_rows.get_value()
{
if *value > 1 {
- self.input()
- .partition_statistics(None)?
- .num_rows
- .to_inexact()
+ child_statistics.num_rows.to_inexact()
} else if *value == 0 {
- // Aggregation on an empty table creates a null row.
- self.input()
- .partition_statistics(None)?
- .num_rows
- .add(&Precision::Exact(1))
+ child_statistics.num_rows
} else {
// num_rows = 1 case
- self.input().partition_statistics(None)?.num_rows
+ let grouping_set_num = self.group_by.groups.len();
+ child_statistics.num_rows.map(|x| x * grouping_set_num)
Review Comment:
I don't get it here, could you please give a SQL example?
##########
datafusion/physical-plan/src/aggregates/mod.rs:
##########
@@ -733,13 +733,33 @@ impl AggregateExec {
&self.input_order_mode
}
- fn statistics_inner(&self) -> Result<Statistics> {
+ fn statistics_inner(&self, child_statistics: Statistics) -> Result<Statistics> {
// TODO stats: group expressions:
// - once expressions will be able to compute their own stats, use it here
// - case where we group by on a column for which with have the `distinct` stat
// TODO stats: aggr expression:
// - aggregations sometimes also preserve invariants such as min, max...
- let column_statistics = Statistics::unknown_column(&self.schema());
+
+ let column_statistics = {
+ // self.schema: [<group by exprs>, <aggregate exprs>]
+ let mut column_statistics = Statistics::unknown_column(&self.schema());
+
+ for (idx, (expr, _)) in self.group_by.expr.iter().enumerate() {
Review Comment:
If the child of the agg node is a non-agg node, do the output columns of the child match with the `group by expr`?
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] implement `AggregateExec.partition_statistics` [datafusion]
Posted by "alamb (via GitHub)" <gi...@apache.org>.
alamb commented on PR #15954:
URL: https://github.com/apache/datafusion/pull/15954#issuecomment-2916893985
🎉
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] implement `AggregateExec.partition_statistics` [datafusion]
Posted by "UBarney (via GitHub)" <gi...@apache.org>.
UBarney commented on PR #15954:
URL: https://github.com/apache/datafusion/pull/15954#issuecomment-2856940325
FYI @xudong963
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] implement `AggregateExec.partition_statistics` [datafusion]
Posted by "UBarney (via GitHub)" <gi...@apache.org>.
UBarney commented on code in PR #15954:
URL: https://github.com/apache/datafusion/pull/15954#discussion_r2074906893
##########
datafusion/physical-plan/src/aggregates/mod.rs:
##########
@@ -751,28 +771,16 @@ impl AggregateExec {
})
}
_ => {
- // When the input row count is 0 or 1, we can adopt that statistic keeping its reliability.
+ // When the input row count is 1, we can adopt that statistic keeping its reliability.
// When it is larger than 1, we degrade the precision since it may decrease after aggregation.
- let num_rows = if let Some(value) = self
- .input()
- .partition_statistics(None)?
- .num_rows
- .get_value()
+ let num_rows = if let Some(value) = child_statistics.num_rows.get_value()
{
- if *value > 1 {
- self.input()
- .partition_statistics(None)?
- .num_rows
- .to_inexact()
- } else if *value == 0 {
- // Aggregation on an empty table creates a null row.
Review Comment:
* If `!group_by_expr.is_empty()` and `input_statistics.num_rows == 0`:
* Both `Partial` and `Final` aggregation modes (`agg.mode`) correctly yield 0 output rows.
* I added `dbg!(&output_batch.num_rows())` before [this line](https://github.com/apache/datafusion/blob/9d2f04996604e709ee440b65f41e7b882f50b788/datafusion/physical-plan/src/aggregates/row_hash.rs#L800). It printed nothing, since `self.exec_state` likely transitioned to `ExecutionState::Done`.
* If `group_by_expr.is_empty()` and `input_statistics.num_rows == 0`:
* `Final` aggregation mode (`agg.mode == Final`) yields 1 output row
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] implement `AggregateExec.partition_statistics` [datafusion]
Posted by "xudong963 (via GitHub)" <gi...@apache.org>.
xudong963 commented on code in PR #15954:
URL: https://github.com/apache/datafusion/pull/15954#discussion_r2081997257
##########
datafusion/physical-plan/src/aggregates/mod.rs:
##########
@@ -3041,4 +3048,96 @@ mod tests {
run_test_with_spill_pool_if_necessary(20_000, false).await?;
Ok(())
}
+
+ #[tokio::test]
+ async fn test_statistic() -> Result<()> {
Review Comment:
I also suggest moving the test into one place like other plans: https://github.com/apache/datafusion/blob/main/datafusion/core/tests/physical_optimizer/partition_statistics.rs
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] implement `AggregateExec.partition_statistics` [datafusion]
Posted by "UBarney (via GitHub)" <gi...@apache.org>.
UBarney commented on code in PR #15954:
URL: https://github.com/apache/datafusion/pull/15954#discussion_r2074906893
##########
datafusion/physical-plan/src/aggregates/mod.rs:
##########
@@ -751,28 +771,16 @@ impl AggregateExec {
})
}
_ => {
- // When the input row count is 0 or 1, we can adopt that statistic keeping its reliability.
+ // When the input row count is 1, we can adopt that statistic keeping its reliability.
// When it is larger than 1, we degrade the precision since it may decrease after aggregation.
- let num_rows = if let Some(value) = self
- .input()
- .partition_statistics(None)?
- .num_rows
- .get_value()
+ let num_rows = if let Some(value) = child_statistics.num_rows.get_value()
{
- if *value > 1 {
- self.input()
- .partition_statistics(None)?
- .num_rows
- .to_inexact()
- } else if *value == 0 {
- // Aggregation on an empty table creates a null row.
Review Comment:
* If `!group_by_expr.is_empty()` and `input_statistics.num_rows == 0`:
* Both `Partial` and `Final` aggregation modes (`agg.mode`) yield 0 output rows.
```
> select count(*) from generate_series(0) where value > 10 group by value;
+----------+
| count(*) |
+----------+
+----------+
0 row(s) fetched.
```
* (I added `dbg!(&output_batch.num_rows())` before [this line](https://github.com/apache/datafusion/blob/9d2f04996604e709ee440b65f41e7b882f50b788/datafusion/physical-plan/src/aggregates/row_hash.rs#L800). It printed nothing, since `self.exec_state` likely transitioned to `ExecutionState::Done`.)
* If `group_by_expr.is_empty()` and `input_statistics.num_rows == 0`:
* `Final` aggregation mode (`agg.mode == Final`) yields 1 output row. But it already return never hit this line
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] implement `AggregateExec.partition_statistics` [datafusion]
Posted by "berkaysynnada (via GitHub)" <gi...@apache.org>.
berkaysynnada merged PR #15954:
URL: https://github.com/apache/datafusion/pull/15954
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] implement `AggregateExec.partition_statistics` [datafusion]
Posted by "UBarney (via GitHub)" <gi...@apache.org>.
UBarney commented on code in PR #15954:
URL: https://github.com/apache/datafusion/pull/15954#discussion_r2105771450
##########
datafusion/physical-plan/src/aggregates/mod.rs:
##########
@@ -733,13 +733,33 @@ impl AggregateExec {
&self.input_order_mode
}
- fn statistics_inner(&self) -> Result<Statistics> {
+ fn statistics_inner(&self, child_statistics: Statistics) -> Result<Statistics> {
// TODO stats: group expressions:
// - once expressions will be able to compute their own stats, use it here
// - case where we group by on a column for which with have the `distinct` stat
// TODO stats: aggr expression:
// - aggregations sometimes also preserve invariants such as min, max...
- let column_statistics = Statistics::unknown_column(&self.schema());
+
+ let column_statistics = {
+ // self.schema: [<group by exprs>, <aggregate exprs>]
+ let mut column_statistics = Statistics::unknown_column(&self.schema());
+
+ for (idx, (expr, _)) in self.group_by.expr.iter().enumerate() {
Review Comment:
> If we adjust the order of group by exprs in the test, does it still match?
Yes.
<details>
#[tokio::test]
async fn test_agg_order() -> Result<()> {
let scan = create_scan_exec_with_statistics(None, Some(2)).await;
let scan_schema = scan.schema();
let group_by = PhysicalGroupBy::new_single(vec![
(
binary(
lit(1),
Operator::Plus,
col("id", &scan_schema)?,
&scan_schema,
)?,
"expr".to_string(),
),
(col("id", &scan_schema)?, "id".to_string()),
]);
let aggr_expr = vec![AggregateExprBuilder::new(count_udaf(), vec![lit(1)])
.schema(Arc::clone(&scan_schema))
.alias(String::from("COUNT(c)"))
.build()
.map(Arc::new)?];
let aggregate_exec_partial = Arc::new(AggregateExec::try_new(
AggregateMode::Partial,
group_by.clone(),
aggr_expr.clone(),
vec![None],
Arc::clone(&scan),
scan_schema.clone(),
)?);
let p0_statistics = aggregate_exec_partial.partition_statistics(Some(0))?;
let expected_p0_statistics = Statistics {
num_rows: Precision::Inexact(2),
total_byte_size: Precision::Absent,
column_statistics: vec![
ColumnStatistics::new_unknown(),
ColumnStatistics {
null_count: Precision::Absent,
max_value: Precision::Exact(ScalarValue::Int32(Some(4))),
min_value: Precision::Exact(ScalarValue::Int32(Some(3))),
sum_value: Precision::Absent,
distinct_count: Precision::Absent,
},
ColumnStatistics::new_unknown(),
],
};
assert_eq!(&p0_statistics, &expected_p0_statistics);
let expected_p1_statistics = Statistics {
num_rows: Precision::Inexact(2),
total_byte_size: Precision::Absent,
column_statistics: vec![
ColumnStatistics::new_unknown(),
ColumnStatistics {
null_count: Precision::Absent,
max_value: Precision::Exact(ScalarValue::Int32(Some(2))),
min_value: Precision::Exact(ScalarValue::Int32(Some(1))),
sum_value: Precision::Absent,
distinct_count: Precision::Absent,
},
ColumnStatistics::new_unknown(),
],
};
let p1_statistics = aggregate_exec_partial.partition_statistics(Some(1))?;
assert_eq!(&p1_statistics, &expected_p1_statistics);
validate_statistics_with_data(
aggregate_exec_partial.clone(),
vec![
ExpectedStatistics::NonEmpty(3, 4, 2),
ExpectedStatistics::NonEmpty(1, 2, 2),
],
1,
)
.await?;
let agg_final = Arc::new(AggregateExec::try_new(
AggregateMode::FinalPartitioned,
group_by.as_final(),
aggr_expr.clone(),
vec![None],
aggregate_exec_partial.clone(),
aggregate_exec_partial.schema(),
)?);
let p0_statistics = agg_final.partition_statistics(Some(0))?;
assert_eq!(&p0_statistics, &expected_p0_statistics);
let p1_statistics = agg_final.partition_statistics(Some(1))?;
assert_eq!(&p1_statistics, &expected_p1_statistics);
validate_statistics_with_data(
agg_final.clone(),
vec![
ExpectedStatistics::NonEmpty(3, 4, 2),
ExpectedStatistics::NonEmpty(1, 2, 2),
],
1,
)
.await?;
Ok(())
}
</details>
> The idx will be 1
`col.index()` will stay same (`0`) https://github.com/apache/datafusion/blob/2d801940c3cb0cec3209aa890688590ded791865/datafusion/physical-expr/src/expressions/column.rs#L167-L169
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] implement `AggregateExec.partition_statistics` [datafusion]
Posted by "UBarney (via GitHub)" <gi...@apache.org>.
UBarney commented on code in PR #15954:
URL: https://github.com/apache/datafusion/pull/15954#discussion_r2102533299
##########
datafusion/physical-plan/src/aggregates/mod.rs:
##########
@@ -751,28 +771,16 @@ impl AggregateExec {
})
}
_ => {
- // When the input row count is 0 or 1, we can adopt that statistic keeping its reliability.
+ // When the input row count is 1, we can adopt that statistic keeping its reliability.
// When it is larger than 1, we degrade the precision since it may decrease after aggregation.
- let num_rows = if let Some(value) = self
- .input()
- .partition_statistics(None)?
- .num_rows
- .get_value()
+ let num_rows = if let Some(value) = child_statistics.num_rows.get_value()
{
- if *value > 1 {
- self.input()
- .partition_statistics(None)?
- .num_rows
- .to_inexact()
- } else if *value == 0 {
- // Aggregation on an empty table creates a null row.
Review Comment:
~~I think current unit test already covered it~~
~~+ [`group_by_expr.is_empty() and input_statistics.num_rows == 0`](https://github.com/apache/datafusion/blob/0c7760aebd4e057196b363a173a9884bcdbb7236/datafusion/core/tests/physical_optimizer/partition_statistics.rs#L673-L708)~~
~~+ [`!group_by_expr.is_empty() and input_statistics.num_rows == 0`](https://github.com/apache/datafusion/blob/0c7760aebd4e057196b363a173a9884bcdbb7236/datafusion/core/tests/physical_optimizer/partition_statistics.rs#L632-L663)~~~
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] implement `AggregateExec.partition_statistics` [datafusion]
Posted by "xudong963 (via GitHub)" <gi...@apache.org>.
xudong963 commented on code in PR #15954:
URL: https://github.com/apache/datafusion/pull/15954#discussion_r2105705916
##########
datafusion/physical-plan/src/aggregates/mod.rs:
##########
@@ -751,28 +771,18 @@ impl AggregateExec {
})
}
_ => {
- // When the input row count is 0 or 1, we can adopt that statistic keeping its reliability.
+ // When the input row count is 1, we can adopt that statistic keeping its reliability.
// When it is larger than 1, we degrade the precision since it may decrease after aggregation.
- let num_rows = if let Some(value) = self
- .input()
- .partition_statistics(None)?
- .num_rows
- .get_value()
+ let num_rows = if let Some(value) = child_statistics.num_rows.get_value()
{
if *value > 1 {
- self.input()
- .partition_statistics(None)?
- .num_rows
- .to_inexact()
+ child_statistics.num_rows.to_inexact()
} else if *value == 0 {
- // Aggregation on an empty table creates a null row.
- self.input()
- .partition_statistics(None)?
- .num_rows
- .add(&Precision::Exact(1))
+ child_statistics.num_rows
} else {
// num_rows = 1 case
- self.input().partition_statistics(None)?.num_rows
+ let grouping_set_num = self.group_by.groups.len();
+ child_statistics.num_rows.map(|x| x * grouping_set_num)
Review Comment:
I think it's also a bug fix, do we have a ut?
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] implement `AggregateExec.partition_statistics` [datafusion]
Posted by "xudong963 (via GitHub)" <gi...@apache.org>.
xudong963 commented on code in PR #15954:
URL: https://github.com/apache/datafusion/pull/15954#discussion_r2081995641
##########
datafusion/physical-plan/src/aggregates/mod.rs:
##########
@@ -3041,4 +3048,96 @@ mod tests {
run_test_with_spill_pool_if_necessary(20_000, false).await?;
Ok(())
}
+
+ #[tokio::test]
+ async fn test_statistic() -> Result<()> {
Review Comment:
It's better to have the real run statistics, see here: https://github.com/apache/datafusion/blob/main/datafusion/core/tests/physical_optimizer/partition_statistics.rs#L233-L235
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] implement `AggregateExec.partition_statistics` [datafusion]
Posted by "UBarney (via GitHub)" <gi...@apache.org>.
UBarney commented on code in PR #15954:
URL: https://github.com/apache/datafusion/pull/15954#discussion_r2074906893
##########
datafusion/physical-plan/src/aggregates/mod.rs:
##########
@@ -751,28 +771,16 @@ impl AggregateExec {
})
}
_ => {
- // When the input row count is 0 or 1, we can adopt that statistic keeping its reliability.
+ // When the input row count is 1, we can adopt that statistic keeping its reliability.
// When it is larger than 1, we degrade the precision since it may decrease after aggregation.
- let num_rows = if let Some(value) = self
- .input()
- .partition_statistics(None)?
- .num_rows
- .get_value()
+ let num_rows = if let Some(value) = child_statistics.num_rows.get_value()
{
- if *value > 1 {
- self.input()
- .partition_statistics(None)?
- .num_rows
- .to_inexact()
- } else if *value == 0 {
- // Aggregation on an empty table creates a null row.
Review Comment:
* If `!group_by_expr.is_empty()` and `input_statistics.num_rows == 0`:
* Both `Partial` and `Final` aggregation modes (`agg.mode`) yield 0 output rows.
* (I added `dbg!(&output_batch.num_rows())` before [this line](https://github.com/apache/datafusion/blob/9d2f04996604e709ee440b65f41e7b882f50b788/datafusion/physical-plan/src/aggregates/row_hash.rs#L800). It printed nothing, since `self.exec_state` likely transitioned to `ExecutionState::Done`.)
* If `group_by_expr.is_empty()` and `input_statistics.num_rows == 0`:
* `Final` aggregation mode (`agg.mode == Final`) yields 1 output row. But it already return never hit this line
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] implement `AggregateExec.partition_statistics` [datafusion]
Posted by "xudong963 (via GitHub)" <gi...@apache.org>.
xudong963 commented on PR #15954:
URL: https://github.com/apache/datafusion/pull/15954#issuecomment-2916884677
@UBarney Thank you again!!
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] implement `AggregateExec.partition_statistics` [datafusion]
Posted by "UBarney (via GitHub)" <gi...@apache.org>.
UBarney commented on code in PR #15954:
URL: https://github.com/apache/datafusion/pull/15954#discussion_r2105771450
##########
datafusion/physical-plan/src/aggregates/mod.rs:
##########
@@ -733,13 +733,33 @@ impl AggregateExec {
&self.input_order_mode
}
- fn statistics_inner(&self) -> Result<Statistics> {
+ fn statistics_inner(&self, child_statistics: Statistics) -> Result<Statistics> {
// TODO stats: group expressions:
// - once expressions will be able to compute their own stats, use it here
// - case where we group by on a column for which with have the `distinct` stat
// TODO stats: aggr expression:
// - aggregations sometimes also preserve invariants such as min, max...
- let column_statistics = Statistics::unknown_column(&self.schema());
+
+ let column_statistics = {
+ // self.schema: [<group by exprs>, <aggregate exprs>]
+ let mut column_statistics = Statistics::unknown_column(&self.schema());
+
+ for (idx, (expr, _)) in self.group_by.expr.iter().enumerate() {
Review Comment:
> If we adjust the order of group by exprs in the test, does it still match?
Yes.
<details>
#[tokio::test]
async fn test_agg_order() -> Result<()> {
let scan = create_scan_exec_with_statistics(None, Some(2)).await;
let scan_schema = scan.schema();
let group_by = PhysicalGroupBy::new_single(vec![
(
binary(
lit(1),
Operator::Plus,
col("id", &scan_schema)?,
&scan_schema,
)?,
"expr".to_string(),
),
(col("id", &scan_schema)?, "id".to_string()),
]);
let aggr_expr = vec![AggregateExprBuilder::new(count_udaf(), vec![lit(1)])
.schema(Arc::clone(&scan_schema))
.alias(String::from("COUNT(c)"))
.build()
.map(Arc::new)?];
let aggregate_exec_partial = Arc::new(AggregateExec::try_new(
AggregateMode::Partial,
group_by.clone(),
aggr_expr.clone(),
vec![None],
Arc::clone(&scan),
scan_schema.clone(),
)?);
let p0_statistics = aggregate_exec_partial.partition_statistics(Some(0))?;
let expected_p0_statistics = Statistics {
num_rows: Precision::Inexact(2),
total_byte_size: Precision::Absent,
column_statistics: vec![
ColumnStatistics::new_unknown(),
ColumnStatistics {
null_count: Precision::Absent,
max_value: Precision::Exact(ScalarValue::Int32(Some(4))),
min_value: Precision::Exact(ScalarValue::Int32(Some(3))),
sum_value: Precision::Absent,
distinct_count: Precision::Absent,
},
ColumnStatistics::new_unknown(),
],
};
assert_eq!(&p0_statistics, &expected_p0_statistics);
let expected_p1_statistics = Statistics {
num_rows: Precision::Inexact(2),
total_byte_size: Precision::Absent,
column_statistics: vec![
ColumnStatistics::new_unknown(),
ColumnStatistics {
null_count: Precision::Absent,
max_value: Precision::Exact(ScalarValue::Int32(Some(2))),
min_value: Precision::Exact(ScalarValue::Int32(Some(1))),
sum_value: Precision::Absent,
distinct_count: Precision::Absent,
},
ColumnStatistics::new_unknown(),
],
};
let p1_statistics = aggregate_exec_partial.partition_statistics(Some(1))?;
assert_eq!(&p1_statistics, &expected_p1_statistics);
validate_statistics_with_data(
aggregate_exec_partial.clone(),
vec![
ExpectedStatistics::NonEmpty(3, 4, 2),
ExpectedStatistics::NonEmpty(1, 2, 2),
],
1,
)
.await?;
let agg_final = Arc::new(AggregateExec::try_new(
AggregateMode::FinalPartitioned,
group_by.as_final(),
aggr_expr.clone(),
vec![None],
aggregate_exec_partial.clone(),
aggregate_exec_partial.schema(),
)?);
let p0_statistics = agg_final.partition_statistics(Some(0))?;
assert_eq!(&p0_statistics, &expected_p0_statistics);
let p1_statistics = agg_final.partition_statistics(Some(1))?;
assert_eq!(&p1_statistics, &expected_p1_statistics);
validate_statistics_with_data(
agg_final.clone(),
vec![
ExpectedStatistics::NonEmpty(3, 4, 2),
ExpectedStatistics::NonEmpty(1, 2, 2),
],
1,
)
.await?;
Ok(())
}
</details>
> The idx will be 1
Yes. but I use `child_statistics.column_statistics[col.index()]` rather than `child_statistics.column_statistics.[idx]`
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] implement `AggregateExec.partition_statistics` [datafusion]
Posted by "xudong963 (via GitHub)" <gi...@apache.org>.
xudong963 commented on PR #15954:
URL: https://github.com/apache/datafusion/pull/15954#issuecomment-2876248152
> @xudong963 Thanks for reviewing. All comments have been addressed, PTAL
Thank you, I'll review asap
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] implement `AggregateExec.partition_statistics` [datafusion]
Posted by "UBarney (via GitHub)" <gi...@apache.org>.
UBarney commented on code in PR #15954:
URL: https://github.com/apache/datafusion/pull/15954#discussion_r2074906893
##########
datafusion/physical-plan/src/aggregates/mod.rs:
##########
@@ -751,28 +771,16 @@ impl AggregateExec {
})
}
_ => {
- // When the input row count is 0 or 1, we can adopt that statistic keeping its reliability.
+ // When the input row count is 1, we can adopt that statistic keeping its reliability.
// When it is larger than 1, we degrade the precision since it may decrease after aggregation.
- let num_rows = if let Some(value) = self
- .input()
- .partition_statistics(None)?
- .num_rows
- .get_value()
+ let num_rows = if let Some(value) = child_statistics.num_rows.get_value()
{
- if *value > 1 {
- self.input()
- .partition_statistics(None)?
- .num_rows
- .to_inexact()
- } else if *value == 0 {
- // Aggregation on an empty table creates a null row.
Review Comment:
* If `!group_by_expr.is_empty()` and `input_statistics.num_rows == 0`:
* Both `Partial` and `Final` aggregation modes (`agg.mode`) yield 0 output rows. ((Note the AggregateExec metric: `[output_rows=0]`)
```
> explain analyze select count(*) from generate_series(0) where value > 10 group by value;
+-------------------+-----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+
| plan_type | plan |
+-------------------+-----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+
| Plan with Metrics | ProjectionExec: expr=[count(Int64(1))@1 as count(*)], metrics=[output_rows=0, elapsed_compute=24ns] |
| | AggregateExec: mode=FinalPartitioned, gby=[value@0 as value], aggr=[count(Int64(1))], metrics=[output_rows=0, elapsed_compute=100.016µs, spill_count=0, spilled_bytes=0, spilled_rows=0, peak_mem_used=1536] |
| | CoalesceBatchesExec: target_batch_size=8192, metrics=[output_rows=0, elapsed_compute=452ns] |
| | RepartitionExec: partitioning=Hash([value@0], 24), input_partitions=24, metrics=[fetch_time=10.544607ms, repartition_time=24ns, send_time=576ns] |
| | AggregateExec: mode=Partial, gby=[value@0 as value], aggr=[count(Int64(1))], metrics=[output_rows=0, elapsed_compute=170.537µs, spill_count=0, spilled_bytes=0, spilled_rows=0, skipped_aggregation_rows=0, peak_mem_used=1536] |
| | CoalesceBatchesExec: target_batch_size=8192, metrics=[output_rows=0, elapsed_compute=663ns] |
| | FilterExec: value@0 > 10, metrics=[output_rows=0, elapsed_compute=2.201314ms] |
| | RepartitionExec: partitioning=RoundRobinBatch(24), input_partitions=1, metrics=[fetch_time=3.077µs, repartition_time=1ns, send_time=1.16µs] |
| | LazyMemoryExec: partitions=1, batch_generators=[generate_series: start=0, end=0, batch_size=8192], metrics=[] |
| | |
+-------------------+-----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+
1 row(s) fetched.
Elapsed 0.004 seconds.
> select count(*) from generate_series(0) where value > 10 group by value;
+----------+
| count(*) |
+----------+
+----------+
0 row(s) fetched.
```
* If `group_by_expr.is_empty()` and `input_statistics.num_rows == 0`:
* `Final` aggregation mode (`agg.mode == Final`) yields 1 output row. But it already return never hit this line
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] implement `AggregateExec.partition_statistics` [datafusion]
Posted by "UBarney (via GitHub)" <gi...@apache.org>.
UBarney commented on code in PR #15954:
URL: https://github.com/apache/datafusion/pull/15954#discussion_r2104159225
##########
datafusion/physical-plan/src/aggregates/mod.rs:
##########
@@ -751,28 +771,16 @@ impl AggregateExec {
})
}
_ => {
- // When the input row count is 0 or 1, we can adopt that statistic keeping its reliability.
+ // When the input row count is 1, we can adopt that statistic keeping its reliability.
// When it is larger than 1, we degrade the precision since it may decrease after aggregation.
- let num_rows = if let Some(value) = self
- .input()
- .partition_statistics(None)?
- .num_rows
- .get_value()
+ let num_rows = if let Some(value) = child_statistics.num_rows.get_value()
{
- if *value > 1 {
- self.input()
- .partition_statistics(None)?
- .num_rows
- .to_inexact()
- } else if *value == 0 {
- // Aggregation on an empty table creates a null row.
Review Comment:
https://github.com/apache/datafusion/blob/aa62950b7b288efc7eac4754aa54de29a5d28d93/datafusion/core/tests/physical_optimizer/partition_statistics.rs#L636-L662
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] implement `AggregateExec.partition_statistics` [datafusion]
Posted by "UBarney (via GitHub)" <gi...@apache.org>.
UBarney commented on code in PR #15954:
URL: https://github.com/apache/datafusion/pull/15954#discussion_r2102508622
##########
datafusion/physical-plan/src/aggregates/mod.rs:
##########
@@ -751,28 +771,18 @@ impl AggregateExec {
})
}
_ => {
- // When the input row count is 0 or 1, we can adopt that statistic keeping its reliability.
+ // When the input row count is 1, we can adopt that statistic keeping its reliability.
// When it is larger than 1, we degrade the precision since it may decrease after aggregation.
- let num_rows = if let Some(value) = self
- .input()
- .partition_statistics(None)?
- .num_rows
- .get_value()
+ let num_rows = if let Some(value) = child_statistics.num_rows.get_value()
{
if *value > 1 {
- self.input()
- .partition_statistics(None)?
- .num_rows
- .to_inexact()
+ child_statistics.num_rows.to_inexact()
} else if *value == 0 {
- // Aggregation on an empty table creates a null row.
- self.input()
- .partition_statistics(None)?
- .num_rows
- .add(&Precision::Exact(1))
+ child_statistics.num_rows
} else {
// num_rows = 1 case
- self.input().partition_statistics(None)?.num_rows
+ let grouping_set_num = self.group_by.groups.len();
+ child_statistics.num_rows.map(|x| x * grouping_set_num)
Review Comment:
```
> explain analyze select t1.value, t2.value, count(*) from range(1) as t1 join range(1) as t2 using(value) group by cube(t1.value, t2.value);
+-------------------+---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+
| plan_type | plan |
+-------------------+---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+
| Plan with Metrics | ProjectionExec: expr=[value@0 as value, value@1 as value, count(Int64(1))@3 as count(*)], metrics=[output_rows=4, elapsed_compute=20.988µs] |
| | AggregateExec: mode=FinalPartitioned, gby=[value@0 as value, value@1 as value, __grouping_id@2 as __grouping_id], aggr=[count(Int64(1))], metrics=[output_rows=4, elapsed_compute=107.217µs, spill_count=0, spilled_bytes=0, spilled_rows=0, peak_mem_used=3264] |
| | CoalesceBatchesExec: target_batch_size=8192, metrics=[output_rows=4, elapsed_compute=4.21µs] |
| | RepartitionExec: partitioning=Hash([value@0, value@1, __grouping_id@2], 24), input_partitions=24, metrics=[fetch_time=19.129834ms, repartition_time=7.618µs, send_time=35.082µs] |
| | AggregateExec: mode=Partial, gby=[(value@0 as value, value@1 as value), (NULL as value, value@1 as value), (value@0 as value, NULL as value), (NULL as value, NULL as value)], aggr=[count(Int64(1))], metrics=[output_rows=4, elapsed_compute=120.603µs, spill_count=0, spilled_bytes=0, spilled_rows=0, peak_mem_used=1976] |
| | CoalesceBatchesExec: target_batch_size=8192, metrics=[output_rows=1, elapsed_compute=1.288µs] |
| | HashJoinExec: mode=Partitioned, join_type=Inner, on=[(value@0, value@0)], metrics=[output_rows=1, build_input_batches=1, build_input_rows=1, input_batches=1, input_rows=1, output_batches=1, build_mem_used=1760, build_time=475.105µs, join_time=14.496µs] |
| | CoalesceBatchesExec: target_batch_size=8192, metrics=[output_rows=1, elapsed_compute=1.622µs] |
| | RepartitionExec: partitioning=Hash([value@0], 24), input_partitions=24, metrics=[fetch_time=654.947µs, repartition_time=4.621µs, send_time=1.056µs] |
| | RepartitionExec: partitioning=RoundRobinBatch(24), input_partitions=1, metrics=[fetch_time=3.096µs, repartition_time=1ns, send_time=741ns] |
| | LazyMemoryExec: partitions=1, batch_generators=[range: start=0, end=1, batch_size=8192], metrics=[] |
| | CoalesceBatchesExec: target_batch_size=8192, metrics=[output_rows=1, elapsed_compute=2.197µs] |
| | RepartitionExec: partitioning=Hash([value@0], 24), input_partitions=24, metrics=[fetch_time=2.702565ms, repartition_time=2.726µs, send_time=1.497µs] |
| | RepartitionExec: partitioning=RoundRobinBatch(24), input_partitions=1, metrics=[fetch_time=1.571µs, repartition_time=1ns, send_time=248ns] |
| | LazyMemoryExec: partitions=1, batch_generators=[range: start=0, end=1, batch_size=8192], metrics=[] |
| | |
+-------------------+---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+
1 row(s) fetched.
Elapsed 0.004 seconds.
>
```
##########
datafusion/physical-plan/src/aggregates/mod.rs:
##########
@@ -733,13 +733,33 @@ impl AggregateExec {
&self.input_order_mode
}
- fn statistics_inner(&self) -> Result<Statistics> {
+ fn statistics_inner(&self, child_statistics: Statistics) -> Result<Statistics> {
// TODO stats: group expressions:
// - once expressions will be able to compute their own stats, use it here
// - case where we group by on a column for which with have the `distinct` stat
// TODO stats: aggr expression:
// - aggregations sometimes also preserve invariants such as min, max...
- let column_statistics = Statistics::unknown_column(&self.schema());
+
+ let column_statistics = {
+ // self.schema: [<group by exprs>, <aggregate exprs>]
+ let mut column_statistics = Statistics::unknown_column(&self.schema());
+
+ for (idx, (expr, _)) in self.group_by.expr.iter().enumerate() {
Review Comment:
Yes. It can be verified by this testcase : https://github.com/apache/datafusion/blob/0c7760aebd4e057196b363a173a9884bcdbb7236/datafusion/core/tests/physical_optimizer/partition_statistics.rs#L549-L576
##########
datafusion/physical-plan/src/aggregates/mod.rs:
##########
@@ -751,28 +771,16 @@ impl AggregateExec {
})
}
_ => {
- // When the input row count is 0 or 1, we can adopt that statistic keeping its reliability.
+ // When the input row count is 1, we can adopt that statistic keeping its reliability.
// When it is larger than 1, we degrade the precision since it may decrease after aggregation.
- let num_rows = if let Some(value) = self
- .input()
- .partition_statistics(None)?
- .num_rows
- .get_value()
+ let num_rows = if let Some(value) = child_statistics.num_rows.get_value()
{
- if *value > 1 {
- self.input()
- .partition_statistics(None)?
- .num_rows
- .to_inexact()
- } else if *value == 0 {
- // Aggregation on an empty table creates a null row.
Review Comment:
I think current unit test already covered it
+ [`group_by_expr.is_empty() and input_statistics.num_rows == 0`](https://github.com/apache/datafusion/blob/0c7760aebd4e057196b363a173a9884bcdbb7236/datafusion/core/tests/physical_optimizer/partition_statistics.rs#L673-L708)
+ [`!group_by_expr.is_empty() and input_statistics.num_rows == 0`](https://github.com/apache/datafusion/blob/0c7760aebd4e057196b363a173a9884bcdbb7236/datafusion/core/tests/physical_optimizer/partition_statistics.rs#L632-L663)
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] implement `AggregateExec.partition_statistics` [datafusion]
Posted by "xudong963 (via GitHub)" <gi...@apache.org>.
xudong963 commented on code in PR #15954:
URL: https://github.com/apache/datafusion/pull/15954#discussion_r2105707191
##########
datafusion/physical-plan/src/aggregates/mod.rs:
##########
@@ -733,13 +733,33 @@ impl AggregateExec {
&self.input_order_mode
}
- fn statistics_inner(&self) -> Result<Statistics> {
+ fn statistics_inner(&self, child_statistics: Statistics) -> Result<Statistics> {
// TODO stats: group expressions:
// - once expressions will be able to compute their own stats, use it here
// - case where we group by on a column for which with have the `distinct` stat
// TODO stats: aggr expression:
// - aggregations sometimes also preserve invariants such as min, max...
- let column_statistics = Statistics::unknown_column(&self.schema());
+
+ let column_statistics = {
+ // self.schema: [<group by exprs>, <aggregate exprs>]
+ let mut column_statistics = Statistics::unknown_column(&self.schema());
+
+ for (idx, (expr, _)) in self.group_by.expr.iter().enumerate() {
Review Comment:
If we adjust the order of group by exprs in the test, does it still match?
```
let group_by = PhysicalGroupBy::new_single(vec![
(
binary(
lit(1),
Operator::Plus,
col("id", &scan_schema)?,
&scan_schema,
)?,
"expr".to_string(),
),
(col("id", &scan_schema)?, "id".to_string()),
]);
```
The `idx` will be 1, and it'll point to the `date` column in the scan.
Correct me if I missed something
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] implement `AggregateExec.partition_statistics` [datafusion]
Posted by "berkaysynnada (via GitHub)" <gi...@apache.org>.
berkaysynnada commented on PR #15954:
URL: https://github.com/apache/datafusion/pull/15954#issuecomment-2908784234
> LGTM, thank you @UBarney
As noted in the existing TODOs, there could be more special handling needed in both the group-by expressions and the aggregation logic. However, this is clearly an improvement, and others can build on it in future work. A few "good first issues" could be created from this as well.
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] implement `AggregateExec.partition_statistics` [datafusion]
Posted by "UBarney (via GitHub)" <gi...@apache.org>.
UBarney commented on code in PR #15954:
URL: https://github.com/apache/datafusion/pull/15954#discussion_r2093873182
##########
datafusion/core/tests/physical_optimizer/partition_statistics.rs:
##########
@@ -488,4 +520,155 @@ mod test {
assert_eq!(statistics[0], expected_statistic_partition);
Ok(())
}
+
+ #[tokio::test]
+ async fn test_statistic_by_partition_of_agg() -> Result<()> {
+ let scan = create_scan_exec_with_statistics(None, Some(2)).await;
+
+ let scan_schema = scan.schema();
+
+ let group_by = PhysicalGroupBy::new_single(vec![
+ (col("id", &scan_schema)?, "id".to_string()),
+ (
+ binary(
+ lit(1),
+ Operator::Plus,
+ col("id", &scan_schema)?,
+ &scan_schema,
+ )?,
+ "expr".to_string(),
+ ),
+ ]);
+
+ let aggr_expr = vec![AggregateExprBuilder::new(count_udaf(), vec![lit(1)])
+ .schema(Arc::clone(&scan_schema))
+ .alias(String::from("COUNT(c)"))
+ .build()
+ .map(Arc::new)?];
+
+ let aggregate_exec_partial = Arc::new(AggregateExec::try_new(
+ AggregateMode::Partial,
+ group_by.clone(),
+ aggr_expr.clone(),
+ vec![None],
+ Arc::clone(&scan),
+ scan_schema.clone(),
+ )?);
+
+ let p0_statistics = aggregate_exec_partial.partition_statistics(Some(0))?;
+
+ let expected_p0_statistics = Statistics {
+ num_rows: Precision::Inexact(2),
+ total_byte_size: Precision::Absent,
+ column_statistics: vec![
+ ColumnStatistics {
+ null_count: Precision::Absent,
+ max_value: Precision::Exact(ScalarValue::Int32(Some(4))),
+ min_value: Precision::Exact(ScalarValue::Int32(Some(3))),
+ sum_value: Precision::Absent,
+ distinct_count: Precision::Absent,
+ },
+ ColumnStatistics::new_unknown(),
+ ColumnStatistics::new_unknown(),
+ ],
+ };
+
+ assert_eq!(&p0_statistics, &expected_p0_statistics);
+
+ let expected_p1_statistics = Statistics {
+ num_rows: Precision::Inexact(2),
+ total_byte_size: Precision::Absent,
+ column_statistics: vec![
+ ColumnStatistics {
+ null_count: Precision::Absent,
+ max_value: Precision::Exact(ScalarValue::Int32(Some(2))),
+ min_value: Precision::Exact(ScalarValue::Int32(Some(1))),
+ sum_value: Precision::Absent,
+ distinct_count: Precision::Absent,
+ },
+ ColumnStatistics::new_unknown(),
+ ColumnStatistics::new_unknown(),
+ ],
+ };
+
+ let p1_statistics = aggregate_exec_partial.partition_statistics(Some(1))?;
+ assert_eq!(&p1_statistics, &expected_p1_statistics);
+
+ validate_statistics_with_data(
+ aggregate_exec_partial.clone(),
+ vec![
+ ExpectedStatistics::NonEmpty(3, 4, 2),
+ ExpectedStatistics::NonEmpty(1, 2, 2),
+ ],
+ 0,
+ )
+ .await?;
+
+ let agg_final = Arc::new(AggregateExec::try_new(
+ AggregateMode::Final,
+ group_by.clone(),
+ aggr_expr.clone(),
+ vec![None],
+ aggregate_exec_partial.clone(),
+ aggregate_exec_partial.schema(),
+ )?);
+
+ let p0_statistics = agg_final.partition_statistics(Some(0))?;
+ assert_eq!(&p0_statistics, &expected_p0_statistics);
+
+ let p1_statistics = agg_final.partition_statistics(Some(1))?;
+ assert_eq!(&p1_statistics, &expected_p1_statistics);
+
+ validate_statistics_with_data(
+ agg_final.clone(),
+ vec![
+ ExpectedStatistics::NonEmpty(3, 4, 2),
+ ExpectedStatistics::NonEmpty(1, 2, 2),
+ ],
+ 0,
+ )
+ .await?;
+
+ let empty_table =
+ Arc::new(EmptyExec::new(scan_schema.clone()).with_partitions(2));
+
+ let agg_partial = Arc::new(AggregateExec::try_new(
+ AggregateMode::Partial,
+ group_by.clone(),
+ aggr_expr.clone(),
+ vec![None],
+ empty_table.clone(),
+ scan_schema.clone(),
+ )?);
+
+ let agg_final = Arc::new(AggregateExec::try_new(
+ AggregateMode::Final,
+ group_by.clone(),
+ aggr_expr.clone(),
+ vec![None],
+ agg_partial.clone(),
+ agg_partial.schema(),
+ )?);
+
+ let empty_stat = Statistics {
+ num_rows: Precision::Exact(0),
+ total_byte_size: Precision::Absent,
+ column_statistics: vec![
+ ColumnStatistics::new_unknown(),
+ ColumnStatistics::new_unknown(),
+ ColumnStatistics::new_unknown(),
+ ],
+ };
+ assert_eq!(&empty_stat, &agg_final.partition_statistics(Some(0))?);
+ assert_eq!(&empty_stat, &agg_final.partition_statistics(Some(1))?);
Review Comment:
> How about also adding `AggregateMode::Final` test
done. PTAL
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] implement `AggregateExec.partition_statistics` [datafusion]
Posted by "UBarney (via GitHub)" <gi...@apache.org>.
UBarney commented on code in PR #15954:
URL: https://github.com/apache/datafusion/pull/15954#discussion_r2105771450
##########
datafusion/physical-plan/src/aggregates/mod.rs:
##########
@@ -733,13 +733,33 @@ impl AggregateExec {
&self.input_order_mode
}
- fn statistics_inner(&self) -> Result<Statistics> {
+ fn statistics_inner(&self, child_statistics: Statistics) -> Result<Statistics> {
// TODO stats: group expressions:
// - once expressions will be able to compute their own stats, use it here
// - case where we group by on a column for which with have the `distinct` stat
// TODO stats: aggr expression:
// - aggregations sometimes also preserve invariants such as min, max...
- let column_statistics = Statistics::unknown_column(&self.schema());
+
+ let column_statistics = {
+ // self.schema: [<group by exprs>, <aggregate exprs>]
+ let mut column_statistics = Statistics::unknown_column(&self.schema());
+
+ for (idx, (expr, _)) in self.group_by.expr.iter().enumerate() {
Review Comment:
> If we adjust the order of group by exprs in the test, does it still match?
Yes.
<details>
#[tokio::test]
async fn test_agg_order() -> Result<()> {
let scan = create_scan_exec_with_statistics(None, Some(2)).await;
let scan_schema = scan.schema();
let group_by = PhysicalGroupBy::new_single(vec![
(
binary(
lit(1),
Operator::Plus,
col("id", &scan_schema)?,
&scan_schema,
)?,
"expr".to_string(),
),
(col("id", &scan_schema)?, "id".to_string()),
]);
let aggr_expr = vec![AggregateExprBuilder::new(count_udaf(), vec![lit(1)])
.schema(Arc::clone(&scan_schema))
.alias(String::from("COUNT(c)"))
.build()
.map(Arc::new)?];
let aggregate_exec_partial = Arc::new(AggregateExec::try_new(
AggregateMode::Partial,
group_by.clone(),
aggr_expr.clone(),
vec![None],
Arc::clone(&scan),
scan_schema.clone(),
)?);
let p0_statistics = aggregate_exec_partial.partition_statistics(Some(0))?;
let expected_p0_statistics = Statistics {
num_rows: Precision::Inexact(2),
total_byte_size: Precision::Absent,
column_statistics: vec![
ColumnStatistics::new_unknown(),
ColumnStatistics {
null_count: Precision::Absent,
max_value: Precision::Exact(ScalarValue::Int32(Some(4))),
min_value: Precision::Exact(ScalarValue::Int32(Some(3))),
sum_value: Precision::Absent,
distinct_count: Precision::Absent,
},
ColumnStatistics::new_unknown(),
],
};
assert_eq!(&p0_statistics, &expected_p0_statistics);
let expected_p1_statistics = Statistics {
num_rows: Precision::Inexact(2),
total_byte_size: Precision::Absent,
column_statistics: vec![
ColumnStatistics::new_unknown(),
ColumnStatistics {
null_count: Precision::Absent,
max_value: Precision::Exact(ScalarValue::Int32(Some(2))),
min_value: Precision::Exact(ScalarValue::Int32(Some(1))),
sum_value: Precision::Absent,
distinct_count: Precision::Absent,
},
ColumnStatistics::new_unknown(),
],
};
let p1_statistics = aggregate_exec_partial.partition_statistics(Some(1))?;
assert_eq!(&p1_statistics, &expected_p1_statistics);
validate_statistics_with_data(
aggregate_exec_partial.clone(),
vec![
ExpectedStatistics::NonEmpty(3, 4, 2),
ExpectedStatistics::NonEmpty(1, 2, 2),
],
1,
)
.await?;
let agg_final = Arc::new(AggregateExec::try_new(
AggregateMode::FinalPartitioned,
group_by.as_final(),
aggr_expr.clone(),
vec![None],
aggregate_exec_partial.clone(),
aggregate_exec_partial.schema(),
)?);
let p0_statistics = agg_final.partition_statistics(Some(0))?;
assert_eq!(&p0_statistics, &expected_p0_statistics);
let p1_statistics = agg_final.partition_statistics(Some(1))?;
assert_eq!(&p1_statistics, &expected_p1_statistics);
validate_statistics_with_data(
agg_final.clone(),
vec![
ExpectedStatistics::NonEmpty(3, 4, 2),
ExpectedStatistics::NonEmpty(1, 2, 2),
],
1,
)
.await?;
Ok(())
}
</details>
> The idx will be 1
`col("id", &scan_schema)?.index()` will stay same (`0`) https://github.com/apache/datafusion/blob/2d801940c3cb0cec3209aa890688590ded791865/datafusion/physical-expr/src/expressions/column.rs#L167-L169
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] implement `AggregateExec.partition_statistics` [datafusion]
Posted by "berkaysynnada (via GitHub)" <gi...@apache.org>.
berkaysynnada commented on code in PR #15954:
URL: https://github.com/apache/datafusion/pull/15954#discussion_r2105794126
##########
datafusion/core/tests/physical_optimizer/partition_statistics.rs:
##########
@@ -488,4 +520,213 @@ mod test {
assert_eq!(statistics[0], expected_statistic_partition);
Ok(())
}
+
+ #[tokio::test]
+ async fn test_statistic_by_partition_of_agg() -> Result<()> {
+ let scan = create_scan_exec_with_statistics(None, Some(2)).await;
+
+ let scan_schema = scan.schema();
+
+ let group_by = PhysicalGroupBy::new_single(vec![
+ (col("id", &scan_schema)?, "id".to_string()),
+ (
+ binary(
+ lit(1),
+ Operator::Plus,
+ col("id", &scan_schema)?,
+ &scan_schema,
+ )?,
+ "expr".to_string(),
+ ),
+ ]);
+
+ let aggr_expr = vec![AggregateExprBuilder::new(count_udaf(), vec![lit(1)])
+ .schema(Arc::clone(&scan_schema))
+ .alias(String::from("COUNT(c)"))
+ .build()
+ .map(Arc::new)?];
+
+ let aggregate_exec_partial = Arc::new(AggregateExec::try_new(
+ AggregateMode::Partial,
+ group_by.clone(),
+ aggr_expr.clone(),
+ vec![None],
+ Arc::clone(&scan),
+ scan_schema.clone(),
+ )?);
Review Comment:
you can apply the same to the other plans to ease reviews
##########
datafusion/core/tests/physical_optimizer/partition_statistics.rs:
##########
@@ -488,4 +520,213 @@ mod test {
assert_eq!(statistics[0], expected_statistic_partition);
Ok(())
}
+
+ #[tokio::test]
+ async fn test_statistic_by_partition_of_agg() -> Result<()> {
+ let scan = create_scan_exec_with_statistics(None, Some(2)).await;
+
+ let scan_schema = scan.schema();
+
+ let group_by = PhysicalGroupBy::new_single(vec![
+ (col("id", &scan_schema)?, "id".to_string()),
+ (
+ binary(
+ lit(1),
+ Operator::Plus,
+ col("id", &scan_schema)?,
+ &scan_schema,
+ )?,
+ "expr".to_string(),
+ ),
+ ]);
+
+ let aggr_expr = vec![AggregateExprBuilder::new(count_udaf(), vec![lit(1)])
+ .schema(Arc::clone(&scan_schema))
+ .alias(String::from("COUNT(c)"))
+ .build()
+ .map(Arc::new)?];
+
+ let aggregate_exec_partial = Arc::new(AggregateExec::try_new(
+ AggregateMode::Partial,
+ group_by.clone(),
+ aggr_expr.clone(),
+ vec![None],
+ Arc::clone(&scan),
+ scan_schema.clone(),
+ )?);
Review Comment:
```suggestion
)?) as _;
let mut plan_string = get_plan_string(&aggregate_exec_partial);
let _ = plan_string.swap_remove(1);
let expected_plan = vec![
"AggregateExec: mode=Partial, gby=[id@0 as id, 1 + id@0 as expr], aggr=[COUNT(c)]",
//" DataSourceExec: file_groups={2 groups: [[.../datafusion/core/tests/data/test_statistics_per_partition/date=2025-03-01/j5fUeSDQo22oPyPU.parquet, .../datafusion/core/tests/data/test_statistics_per_partition/date=2025-03-02/j5fUeSDQo22oPyPU.parquet], [.../datafusion/core/tests/data/test_statistics_per_partition/date=2025-03-03/j5fUeSDQo22oPyPU.parquet, .../datafusion/core/tests/data/test_statistics_per_partition/date=2025-03-04/j5fUeSDQo22oPyPU.parquet]]}, projection=[id, date], file_type=parquet
];
assert_eq!(plan_string, expected_plan);
```
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] implement `AggregateExec.partition_statistics` [datafusion]
Posted by "UBarney (via GitHub)" <gi...@apache.org>.
UBarney commented on PR #15954:
URL: https://github.com/apache/datafusion/pull/15954#issuecomment-2916643078
> LGTM, thank you @UBarney
@berkaysynnada Thanks for reviewing. I have addressed your comment
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] implement `AggregateExec.partition_statistics` [datafusion]
Posted by "xudong963 (via GitHub)" <gi...@apache.org>.
xudong963 commented on PR #15954:
URL: https://github.com/apache/datafusion/pull/15954#issuecomment-2858784424
@UBarney Thank you, I'll review it in two days
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] implement `AggregateExec.partition_statistics` [datafusion]
Posted by "UBarney (via GitHub)" <gi...@apache.org>.
UBarney commented on PR #15954:
URL: https://github.com/apache/datafusion/pull/15954#issuecomment-2869638528
@xudong963 Thanks for reviewing. All comments have been addressed, PTAL
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org