You are viewing a plain text version of this content. The canonical link for it is here.
Posted to github@datafusion.apache.org by "buraksenn (via GitHub)" <gi...@apache.org> on 2026/03/13 15:35:16 UTC
[PR] Estimate aggregate output rows using existing NDV statistics [datafusion]
buraksenn opened a new pull request, #20926:
URL: https://github.com/apache/datafusion/pull/20926
## Which issue does this PR close?
Part of https://github.com/apache/datafusion/issues/20766
## Rationale for this change
Grouped aggregations currently estimate output rows as input_rows, ignoring available NDV statistics. Spark's AggregateEstimation and Trino's AggregationStatsRule both use NDV products to tighten this estimate. This PR is highly referenced by both.
- [Spark reference](https://github.com/apache/spark/blob/e8d8e6a8d040d26aae9571e968e0c64bda0875dc/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/statsEstimation/AggregateEstimation.scala#L38-L61)
- [Trino reference](https://github.com/trinodb/trino/blob/43c8c3ba8bff814697c5926149ce13b9532f030b/core/trino-main/src/main/java/io/trino/cost/AggregationStatsRule.java#L92-L101)
## What changes are included in this PR?
- Estimate aggregate output rows as min(input_rows, product(NDV_i + null_adj_i) * grouping_sets)
- Cap by Top K limit when active since output row cannot be higher than K
- Propagate distinct_count from child stats to group-by output columns
## Are these changes tested?
Yes existing and new tests that cover different scenarios and edge cases
## Are there any user-facing changes?
No
--
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] Estimate aggregate output rows using existing NDV statistics [datafusion]
Posted by "2010YOUY01 (via GitHub)" <gi...@apache.org>.
2010YOUY01 merged PR #20926:
URL: https://github.com/apache/datafusion/pull/20926
--
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] Estimate aggregate output rows using existing NDV statistics [datafusion]
Posted by "asolimando (via GitHub)" <gi...@apache.org>.
asolimando commented on code in PR #20926:
URL: https://github.com/apache/datafusion/pull/20926#discussion_r2959650262
##########
datafusion/physical-plan/src/aggregates/mod.rs:
##########
@@ -4333,6 +4362,157 @@ mod tests {
Ok(())
}
+ #[test]
+ fn test_aggregate_stats_ndv_zero_column() -> Result<()> {
+ use crate::test::exec::StatisticsExec;
+ use datafusion_common::ColumnStatistics;
+
+ let schema = Arc::new(Schema::new(vec![
+ Field::new("a", DataType::Int32, true),
+ Field::new("b", DataType::Int32, true),
+ ]));
+
+ let input_stats = Statistics {
+ num_rows: Precision::Exact(1_000),
+ total_byte_size: Precision::Inexact(1_000),
+ column_statistics: vec![
+ ColumnStatistics {
+ distinct_count: Precision::Exact(0),
+ null_count: Precision::Exact(1_000),
+ ..ColumnStatistics::new_unknown()
+ },
+ ColumnStatistics {
+ distinct_count: Precision::Exact(50),
+ ..ColumnStatistics::new_unknown()
+ },
+ ],
+ };
+
+ let input = Arc::new(StatisticsExec::new(input_stats, (*schema).clone()))
+ as Arc<dyn ExecutionPlan>;
+
+ let agg = AggregateExec::try_new(
+ AggregateMode::Final,
+ PhysicalGroupBy::new_single(vec![
+ (col("a", &schema)? as Arc<dyn PhysicalExpr>, "a".to_string()),
+ (col("b", &schema)? as Arc<dyn PhysicalExpr>, "b".to_string()),
+ ]),
+ vec![Arc::new(
+ AggregateExprBuilder::new(count_udaf(), vec![col("a", &schema)?])
+ .schema(Arc::clone(&schema))
+ .alias("COUNT(a)")
+ .build()?,
+ )],
+ vec![None],
+ input,
+ Arc::clone(&schema),
+ )?;
+
+ let stats = agg.partition_statistics(None)?;
+ // NDV(a)=0 with nulls => max(0+1, 1)=1, NDV(b)=50 => 1*50=50
+ assert_eq!(
+ stats.num_rows,
+ Precision::Inexact(50),
+ "all-null column should contribute 1 to the product, not 0"
+ );
+
+ Ok(())
+ }
+
+ #[test]
+ fn test_aggregate_stats_absent_num_rows_with_ndv() -> Result<()> {
+ use crate::test::exec::StatisticsExec;
+ use datafusion_common::ColumnStatistics;
+
+ let schema = Arc::new(Schema::new(vec![Field::new("a", DataType::Int32, true)]));
+
+ let input_stats = Statistics {
+ num_rows: Precision::Absent,
+ total_byte_size: Precision::Absent,
+ column_statistics: vec![ColumnStatistics {
+ distinct_count: Precision::Exact(100),
+ ..ColumnStatistics::new_unknown()
+ }],
+ };
+
+ let input = Arc::new(StatisticsExec::new(input_stats, (*schema).clone()))
+ as Arc<dyn ExecutionPlan>;
+
+ let agg = AggregateExec::try_new(
+ AggregateMode::Final,
+ PhysicalGroupBy::new_single(vec![(
+ col("a", &schema)? as Arc<dyn PhysicalExpr>,
+ "a".to_string(),
+ )]),
+ vec![Arc::new(
+ AggregateExprBuilder::new(count_udaf(), vec![col("a", &schema)?])
+ .schema(Arc::clone(&schema))
+ .alias("COUNT(a)")
+ .build()?,
+ )],
+ vec![None],
+ input,
+ Arc::clone(&schema),
+ )?;
+
+ let stats = agg.partition_statistics(None)?;
+ assert_eq!(
+ stats.num_rows,
+ Precision::Inexact(100),
+ "absent num_rows should fall back to NDV estimate"
+ );
+
+ Ok(())
+ }
+
+ #[test]
+ fn test_aggregate_stats_absent_num_rows_with_ndv_and_limit() -> Result<()> {
+ use crate::test::exec::StatisticsExec;
+ use datafusion_common::ColumnStatistics;
+
+ let schema = Arc::new(Schema::new(vec![Field::new("a", DataType::Int32, true)]));
+
+ let input_stats = Statistics {
+ num_rows: Precision::Absent,
+ total_byte_size: Precision::Absent,
+ column_statistics: vec![ColumnStatistics {
+ distinct_count: Precision::Exact(100),
+ ..ColumnStatistics::new_unknown()
+ }],
+ };
+
+ let input = Arc::new(StatisticsExec::new(input_stats, (*schema).clone()))
+ as Arc<dyn ExecutionPlan>;
+
+ let mut agg = AggregateExec::try_new(
+ AggregateMode::Final,
+ PhysicalGroupBy::new_single(vec![(
+ col("a", &schema)? as Arc<dyn PhysicalExpr>,
+ "a".to_string(),
+ )]),
+ vec![Arc::new(
+ AggregateExprBuilder::new(count_udaf(), vec![col("a", &schema)?])
+ .schema(Arc::clone(&schema))
+ .alias("COUNT(a)")
+ .build()?,
+ )],
+ vec![None],
+ input,
+ Arc::clone(&schema),
+ )?;
Review Comment:
Nit: the standalone tests all repeat the same schema/StatisticsExec/AggregateExec construction. A small helper (e.g., `build_test_aggregate(schema, stats, group_by, limit) -> AggregateExec`) would reduce the boilerplate.
It's not much about the repeated code, it's more that tests are harder to read, and it's a pity as they are pretty meaningful.
Additionally, the three tests that only check `num_rows` (`ndv_zero_column`, `absent_num_rows_with_ndv`, `absent_num_rows_with_ndv_and_limit`) could be folded into the existing table-driven `test_aggregate_cardinality_estimation`, which is very readable.
--
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] Estimate aggregate output rows using existing NDV statistics [datafusion]
Posted by "buraksenn (via GitHub)" <gi...@apache.org>.
buraksenn commented on PR #20926:
URL: https://github.com/apache/datafusion/pull/20926#issuecomment-4193139085
I've addresses latest code related comments. Thanks for detailed review @2010YOUY01
--
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] Estimate aggregate output rows using existing NDV statistics [datafusion]
Posted by "buraksenn (via GitHub)" <gi...@apache.org>.
buraksenn commented on PR #20926:
URL: https://github.com/apache/datafusion/pull/20926#issuecomment-4225410586
Btw, there is also https://github.com/apache/datafusion/pull/20904 which addresses similar topic and I could not get any maintainer review after @asolimando's detailed one. It would be great to have some review on that if you've time @xudong963 and @2010YOUY01
--
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] Estimate aggregate output rows using existing NDV statistics [datafusion]
Posted by "buraksenn (via GitHub)" <gi...@apache.org>.
buraksenn commented on PR #20926:
URL: https://github.com/apache/datafusion/pull/20926#issuecomment-4225363169
> @xudong963, @2010YOUY01: thanks again for your reviews and help, here and elsewhere!
>
> For this PR specifically, is there anything missing before merging?
Was this meant for another PR @asolimando? This was merged 3 days ago
--
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] Estimate aggregate output rows using existing NDV statistics [datafusion]
Posted by "buraksenn (via GitHub)" <gi...@apache.org>.
buraksenn commented on code in PR #20926:
URL: https://github.com/apache/datafusion/pull/20926#discussion_r2962448495
##########
datafusion/physical-plan/src/aggregates/mod.rs:
##########
@@ -4333,6 +4362,157 @@ mod tests {
Ok(())
}
+ #[test]
+ fn test_aggregate_stats_ndv_zero_column() -> Result<()> {
+ use crate::test::exec::StatisticsExec;
+ use datafusion_common::ColumnStatistics;
+
+ let schema = Arc::new(Schema::new(vec![
+ Field::new("a", DataType::Int32, true),
+ Field::new("b", DataType::Int32, true),
+ ]));
+
+ let input_stats = Statistics {
+ num_rows: Precision::Exact(1_000),
+ total_byte_size: Precision::Inexact(1_000),
+ column_statistics: vec![
+ ColumnStatistics {
+ distinct_count: Precision::Exact(0),
+ null_count: Precision::Exact(1_000),
+ ..ColumnStatistics::new_unknown()
+ },
+ ColumnStatistics {
+ distinct_count: Precision::Exact(50),
+ ..ColumnStatistics::new_unknown()
+ },
+ ],
+ };
+
+ let input = Arc::new(StatisticsExec::new(input_stats, (*schema).clone()))
+ as Arc<dyn ExecutionPlan>;
+
+ let agg = AggregateExec::try_new(
+ AggregateMode::Final,
+ PhysicalGroupBy::new_single(vec![
+ (col("a", &schema)? as Arc<dyn PhysicalExpr>, "a".to_string()),
+ (col("b", &schema)? as Arc<dyn PhysicalExpr>, "b".to_string()),
+ ]),
+ vec![Arc::new(
+ AggregateExprBuilder::new(count_udaf(), vec![col("a", &schema)?])
+ .schema(Arc::clone(&schema))
+ .alias("COUNT(a)")
+ .build()?,
+ )],
+ vec![None],
+ input,
+ Arc::clone(&schema),
+ )?;
+
+ let stats = agg.partition_statistics(None)?;
+ // NDV(a)=0 with nulls => max(0+1, 1)=1, NDV(b)=50 => 1*50=50
+ assert_eq!(
+ stats.num_rows,
+ Precision::Inexact(50),
+ "all-null column should contribute 1 to the product, not 0"
+ );
+
+ Ok(())
+ }
+
+ #[test]
+ fn test_aggregate_stats_absent_num_rows_with_ndv() -> Result<()> {
+ use crate::test::exec::StatisticsExec;
+ use datafusion_common::ColumnStatistics;
+
+ let schema = Arc::new(Schema::new(vec![Field::new("a", DataType::Int32, true)]));
+
+ let input_stats = Statistics {
+ num_rows: Precision::Absent,
+ total_byte_size: Precision::Absent,
+ column_statistics: vec![ColumnStatistics {
+ distinct_count: Precision::Exact(100),
+ ..ColumnStatistics::new_unknown()
+ }],
+ };
+
+ let input = Arc::new(StatisticsExec::new(input_stats, (*schema).clone()))
+ as Arc<dyn ExecutionPlan>;
+
+ let agg = AggregateExec::try_new(
+ AggregateMode::Final,
+ PhysicalGroupBy::new_single(vec![(
+ col("a", &schema)? as Arc<dyn PhysicalExpr>,
+ "a".to_string(),
+ )]),
+ vec![Arc::new(
+ AggregateExprBuilder::new(count_udaf(), vec![col("a", &schema)?])
+ .schema(Arc::clone(&schema))
+ .alias("COUNT(a)")
+ .build()?,
+ )],
+ vec![None],
+ input,
+ Arc::clone(&schema),
+ )?;
+
+ let stats = agg.partition_statistics(None)?;
+ assert_eq!(
+ stats.num_rows,
+ Precision::Inexact(100),
+ "absent num_rows should fall back to NDV estimate"
+ );
+
+ Ok(())
+ }
+
+ #[test]
+ fn test_aggregate_stats_absent_num_rows_with_ndv_and_limit() -> Result<()> {
+ use crate::test::exec::StatisticsExec;
+ use datafusion_common::ColumnStatistics;
+
+ let schema = Arc::new(Schema::new(vec![Field::new("a", DataType::Int32, true)]));
+
+ let input_stats = Statistics {
+ num_rows: Precision::Absent,
+ total_byte_size: Precision::Absent,
+ column_statistics: vec![ColumnStatistics {
+ distinct_count: Precision::Exact(100),
+ ..ColumnStatistics::new_unknown()
+ }],
+ };
+
+ let input = Arc::new(StatisticsExec::new(input_stats, (*schema).clone()))
+ as Arc<dyn ExecutionPlan>;
+
+ let mut agg = AggregateExec::try_new(
+ AggregateMode::Final,
+ PhysicalGroupBy::new_single(vec![(
+ col("a", &schema)? as Arc<dyn PhysicalExpr>,
+ "a".to_string(),
+ )]),
+ vec![Arc::new(
+ AggregateExprBuilder::new(count_udaf(), vec![col("a", &schema)?])
+ .schema(Arc::clone(&schema))
+ .alias("COUNT(a)")
+ .build()?,
+ )],
+ vec![None],
+ input,
+ Arc::clone(&schema),
+ )?;
Review Comment:
Thanks I've applied this changes yes it consolidated cases and some duplicate code
--
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] Estimate aggregate output rows using existing NDV statistics [datafusion]
Posted by "buraksenn (via GitHub)" <gi...@apache.org>.
buraksenn commented on PR #20926:
URL: https://github.com/apache/datafusion/pull/20926#issuecomment-4070358321
Thanks @asolimando for detailed review. I thought this one was more polished looking at Trino and Spark but I've missed some important points. I'll carefully apply your reviews and adjust implementation early tomorrow
--
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] Estimate aggregate output rows using existing NDV statistics [datafusion]
Posted by "asolimando (via GitHub)" <gi...@apache.org>.
asolimando commented on PR #20926:
URL: https://github.com/apache/datafusion/pull/20926#issuecomment-4168534107
@2010YOUY01, apologies for the direct ping, would you be interesting in taking a look?
Re. the discussion in https://github.com/apache/datafusion/issues/21120#issuecomment-4131266771, we have taken Trino and Spark as reference, for grouping sets we went a bit further than those systems, but I believe enough details are provided for the proposed formula (see https://github.com/apache/datafusion/pull/20926#discussion_r2942442151 for the discussion, which @buraksenn captured in code comments).
Is this matching your suggestion for CBO improvements you shared in https://github.com/apache/datafusion/issues/21120#issuecomment-4131266771?
This is an interesting example as it covers as "porting" of existing statistics propagation code from battle-tested systems, and a reasonable extension, so I am particularly interested in your feedback to calibrate future PRs and reviews.
--
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] Estimate aggregate output rows using existing NDV statistics [datafusion]
Posted by "2010YOUY01 (via GitHub)" <gi...@apache.org>.
2010YOUY01 commented on PR #20926:
URL: https://github.com/apache/datafusion/pull/20926#issuecomment-4173976995
> @2010YOUY01, apologies for the direct ping, would you be interesting in taking a look?
>
> Re. the discussion in [#21120 (comment)](https://github.com/apache/datafusion/issues/21120#issuecomment-4131266771), we have taken Trino and Spark as reference, for grouping sets we went a bit further than those systems, but I believe enough details are provided for the proposed formula (see [#20926 (comment)](https://github.com/apache/datafusion/pull/20926#discussion_r2942442151) for the discussion, which @buraksenn captured in code comments).
>
> Is this matching your suggestion for CBO improvements you shared in [#21120 (comment)](https://github.com/apache/datafusion/issues/21120#issuecomment-4131266771)?
>
> This is an interesting example as it covers as "porting" of existing statistics propagation code from battle-tested systems, and a reasonable extension, so I am particularly interested in your feedback to calibrate future PRs and reviews.
Thanks for sharing the context! happy to take a look.
I don't have prior experience on CBO, and I'm still working through the relevant material. Here are several questions:
- why we assume independence between grouping keys — is this just a simplifying heuristic, or do reference systems intentionally do this because overestimating NDV leads to better end-to-end performance (e.g., for join reordering)?
- why we explicitly account for nulls — this adds implementation complexity, but seems to have only a minor impact
### Regarding reference system approach
If we follow a reference-system approach, I suggest going a bit further: identify the best-performing system and try to port its related components more holistically. It’s possible that one system works significantly better than another, and that its cardinality estimation is co-designed with other components like the cost model. Taking ideas from multiple reference systems may not yield good results.
The metrics for deciding reference system IMO is:
1. Simple to implement
2. With comprehensive documentations, explainable
3. Overall improvement on benchmarks like tpcds/join-order-benchmark
DuckDB seems to perform well — they have a thesis (https://blobs.duckdb.org/papers/tom-ebergen-msc-thesis-join-order-optimization-with-almost-no-statistics.pdf) describing their approach, and they’ve reported benchmark results somewhere. Perhaps we can compare it with other reference systems like Spark/Trino similarly, and then pick one reference system and stick to it.
--
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] Estimate aggregate output rows using existing NDV statistics [datafusion]
Posted by "asolimando (via GitHub)" <gi...@apache.org>.
asolimando commented on code in PR #20926:
URL: https://github.com/apache/datafusion/pull/20926#discussion_r2942455195
##########
datafusion/physical-plan/src/aggregates/mod.rs:
##########
@@ -1083,14 +1097,32 @@ impl AggregateExec {
let num_rows = if let Some(value) = child_statistics.num_rows.get_value()
{
if *value > 1 {
- child_statistics.num_rows.to_inexact()
+ let mut num_rows = child_statistics.num_rows.to_inexact();
+
+ if !self.group_by.expr.is_empty() {
+ let ndv_product = self.compute_group_ndv(child_statistics);
+ if let Some(ndv) = ndv_product {
+ let grouping_set_num = self.group_by.groups.len();
+ let ndv_estimate = ndv.saturating_mul(grouping_set_num);
+ num_rows = num_rows.map(|n| n.min(ndv_estimate));
+ }
+ }
+
+ // If TopK mode is active, cap output rows by the limit
+ if let Some(limit_opts) = &self.limit_options {
+ num_rows = num_rows.map(|n| n.min(limit_opts.limit));
+ }
+
+ num_rows
} else if *value == 0 {
child_statistics.num_rows
} else {
// num_rows = 1 case
let grouping_set_num = self.group_by.groups.len();
child_statistics.num_rows.map(|x| x * grouping_set_num)
}
+ } else if let Some(limit_opts) = &self.limit_options {
+ Precision::Inexact(limit_opts.limit)
Review Comment:
Here we might have `ndv` even if `num_rows` is unset, maybe we can return `Inexact(min(ndv, limit))`?
##########
datafusion/physical-plan/src/aggregates/mod.rs:
##########
@@ -1083,14 +1097,32 @@ impl AggregateExec {
let num_rows = if let Some(value) = child_statistics.num_rows.get_value()
{
if *value > 1 {
- child_statistics.num_rows.to_inexact()
+ let mut num_rows = child_statistics.num_rows.to_inexact();
+
+ if !self.group_by.expr.is_empty() {
+ let ndv_product = self.compute_group_ndv(child_statistics);
+ if let Some(ndv) = ndv_product {
+ let grouping_set_num = self.group_by.groups.len();
+ let ndv_estimate = ndv.saturating_mul(grouping_set_num);
Review Comment:
This doesn't look correct to me, grouping sets target different columns.
Consider `GROUPING SETS ((a), (b), (a, b))`, here you would compute `NDV(a) * NDV(b) * 3`, while the number of distinct values should be `NDV(a) + NDV(b) + NDV(a)*NDV(b)`.
[Trino](https://github.com/trinodb/trino/blob/43c8c3ba8bff814697c5926149ce13b9532f030b/core/trino-main/src/main/java/io/trino/cost/AggregationStatsRule.java#L53) is bailing out for (multiple) grouping sets, [Spark](https://github.com/apache/spark/blob/e8d8e6a8d040d26aae9571e968e0c64bda0875dc/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/statsEstimation/AggregateEstimation.scala#L38-L61) does not take them into account as, AFAIU, it rewrites them as union of aggregates at earlier phases.
It's fine to either bail out like Trino does, but if you want to support this, the code would be something like this (pseudocode):
```
res = 0;
for each grouping set gs:
part_res = 0;
for each col in gs:
part_res *= ndv(col) + null_count(col(gs)) > 1 ? 1 : 0;
res += part_res;
```
##########
datafusion/physical-plan/src/aggregates/mod.rs:
##########
@@ -1113,6 +1145,24 @@ impl AggregateExec {
}
}
+ /// Computes `product(NDV_i + null_adjustment_i)` across all group-by columns.
+ /// Returns `None` if any group-by column is not a direct column reference
+ /// or lacks `distinct_count` stats.
+ fn compute_group_ndv(&self, child_statistics: &Statistics) -> Option<usize> {
+ let mut product: usize = 1;
+ for (expr, _) in self.group_by.expr.iter() {
+ let col = expr.as_any().downcast_ref::<Column>()?;
+ let col_stats = &child_statistics.column_statistics[col.index()];
+ let ndv = *col_stats.distinct_count.get_value()?;
Review Comment:
Since we multiply by ndv, we might end up with a total of zero if any ndv is zero. If the column has only null values, you might have `num_rows > = 1` and `ndv = 0`, so let's use `min(num_rows, ndv)` here to be more robust.
##########
datafusion/physical-plan/src/aggregates/mod.rs:
##########
@@ -1113,6 +1145,24 @@ impl AggregateExec {
}
}
+ /// Computes `product(NDV_i + null_adjustment_i)` across all group-by columns.
+ /// Returns `None` if any group-by column is not a direct column reference
+ /// or lacks `distinct_count` stats.
+ fn compute_group_ndv(&self, child_statistics: &Statistics) -> Option<usize> {
+ let mut product: usize = 1;
+ for (expr, _) in self.group_by.expr.iter() {
+ let col = expr.as_any().downcast_ref::<Column>()?;
+ let col_stats = &child_statistics.column_statistics[col.index()];
+ let ndv = *col_stats.distinct_count.get_value()?;
+ let null_adjustment = match col_stats.null_count.get_value() {
+ Some(&n) if n > 0 => 1usize,
+ _ => 0,
Review Comment:
Nit: it's a reasonable default but I'd add this to the comment of the function explicitly
--
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] Estimate aggregate output rows using existing NDV statistics [datafusion]
Posted by "2010YOUY01 (via GitHub)" <gi...@apache.org>.
2010YOUY01 commented on PR #20926:
URL: https://github.com/apache/datafusion/pull/20926#issuecomment-4203083163
Great! Thanks everyone.
--
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] Estimate aggregate output rows using existing NDV statistics [datafusion]
Posted by "buraksenn (via GitHub)" <gi...@apache.org>.
buraksenn commented on code in PR #20926:
URL: https://github.com/apache/datafusion/pull/20926#discussion_r3040198754
##########
datafusion/physical-plan/src/aggregates/mod.rs:
##########
@@ -1113,6 +1163,34 @@ impl AggregateExec {
}
}
+ /// Computes the estimated number of distinct groups across all grouping sets.
+ /// For each grouping set, computes `product(NDV_i + null_adj_i)` for active columns,
+ /// then sums across all sets. Returns `None` if any active column is not a direct
+ /// column reference or lacks `distinct_count` stats.
+ /// When `null_count` is absent or unknown, null_adjustment defaults to 0.
+ fn compute_group_ndv(&self, child_statistics: &Statistics) -> Option<usize> {
Review Comment:
I think this makes sense but to not overcomplicate compute_group_ndv I've extracted `estimate_num_rows` from the caller side to wrap this. Hope it is ok
--
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] Estimate aggregate output rows using existing NDV statistics [datafusion]
Posted by "2010YOUY01 (via GitHub)" <gi...@apache.org>.
2010YOUY01 commented on PR #20926:
URL: https://github.com/apache/datafusion/pull/20926#issuecomment-4188044669
@asolimando Thanks for the detailed and thoughtful reply — really helpful and gave me a lot to think about.
> If a better cardinality estimation (closer to real numbers we can observe at runtime) translates in worse plans, I think it's the cost-model that should be refined. For this reason I find #20292 very compelling for statistics, as it compares estimates vs real quantities, without the noise that the cost model could be adding.
I’ve gone through some of those materials. This might sound a bit pessimistic, but I feel that cardinality estimation is fundamentally hard to get right, and making it mostly correct requires significant engineering effort (which may be challenging for DataFusion at the moment).
Given that, I think it makes sense to start with simple approaches and look for mechanisms outside of CE to avoid disastrous plans.
For example, in this PR’s aggregation estimation, there are two straightforward heuristics:
For estimating the output of `SELECT * FROM t GROUP BY a, b`:
1. Assume independence between `a` and `b`, i.e., `NDV(a) * NDV(b)` — this may overestimate.
2. Assume full dependency, i.e., `max(NDV(a), NDV(b))`.
I imagine a reference system might consistently favor one direction (e.g., always overestimate) and rely on other mechanisms (such as safeguards in the executor) to avoid bad plans.
If we can identify such a reference system, that would be ideal. Otherwise, starting with simple heuristics seems like a reasonable first step. One thing I’d like to avoid is mixing ideas from multiple systems without a consistent set of assumptions, as they may be based on fundamentally different design choices.
--
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] Estimate aggregate output rows using existing NDV statistics [datafusion]
Posted by "2010YOUY01 (via GitHub)" <gi...@apache.org>.
2010YOUY01 commented on code in PR #20926:
URL: https://github.com/apache/datafusion/pull/20926#discussion_r3036219189
##########
datafusion/physical-plan/src/aggregates/mod.rs:
##########
@@ -1037,6 +1037,34 @@ impl AggregateExec {
&self.input_order_mode
}
+ /// Estimates output statistics for this aggregate node.
+ ///
+ /// For grouped aggregations with known input row count > 1, the output row
+ /// count is estimated as:
+ ///
+ /// ```text
+ /// ndv = sum over each grouping set of product(max(NDV_i + nulls_i, 1))
+ /// output_rows = input_rows // baseline
+ /// output_rows = min(output_rows, ndv) // if NDV available
+ /// output_rows = min(output_rows, limit) // if TopK active
+ /// ```
Review Comment:
It would be great to include one or two numeric examples here.
##########
datafusion/physical-plan/src/aggregates/mod.rs:
##########
@@ -1113,6 +1163,34 @@ impl AggregateExec {
}
}
+ /// Computes the estimated number of distinct groups across all grouping sets.
+ /// For each grouping set, computes `product(NDV_i + null_adj_i)` for active columns,
+ /// then sums across all sets. Returns `None` if any active column is not a direct
+ /// column reference or lacks `distinct_count` stats.
+ /// When `null_count` is absent or unknown, null_adjustment defaults to 0.
+ fn compute_group_ndv(&self, child_statistics: &Statistics) -> Option<usize> {
Review Comment:
One minor code structure suggestion, totally optionally.
Now this function only handles ndv estimation, and the fallback logic and limit handling (if returns None, uses num_rows instead) is handled at the caller. The issue is the similar logic gets scattered to multiple places, and this might bring maintenance overhead.
An alternative I think can make it organize nicer: put the fallback logic also inside this function (adding num_rows, limit to arg, and directly return `usize`)
##########
datafusion/physical-plan/src/aggregates/mod.rs:
##########
@@ -1113,6 +1163,34 @@ impl AggregateExec {
}
}
+ /// Computes the estimated number of distinct groups across all grouping sets.
+ /// For each grouping set, computes `product(NDV_i + null_adj_i)` for active columns,
+ /// then sums across all sets. Returns `None` if any active column is not a direct
+ /// column reference or lacks `distinct_count` stats.
+ /// When `null_count` is absent or unknown, null_adjustment defaults to 0.
+ fn compute_group_ndv(&self, child_statistics: &Statistics) -> Option<usize> {
+ let mut total: usize = 0;
+ for group_mask in &self.group_by.groups {
+ let mut set_product: usize = 1;
+ for (j, (expr, _)) in self.group_by.expr.iter().enumerate() {
+ if group_mask[j] {
+ continue;
+ }
+ let col = expr.as_any().downcast_ref::<Column>()?;
Review Comment:
Just to ensure I understand it correctly: If group by expression is non-column (e.g. `abs(a)`), this functions returns None, and its because stats on expressions are not supported yet, and its WIP via https://github.com/apache/datafusion/pull/21122
I think this is also worth mentioning in the comment somewhere.
##########
datafusion/physical-plan/src/aggregates/mod.rs:
##########
@@ -1113,6 +1163,34 @@ impl AggregateExec {
}
}
+ /// Computes the estimated number of distinct groups across all grouping sets.
+ /// For each grouping set, computes `product(NDV_i + null_adj_i)` for active columns,
+ /// then sums across all sets. Returns `None` if any active column is not a direct
+ /// column reference or lacks `distinct_count` stats.
+ /// When `null_count` is absent or unknown, null_adjustment defaults to 0.
Review Comment:
It would be great to add some simple numeric examples for multiple group keys, to demonstrate how is NDV calculated.
--
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] Estimate aggregate output rows using existing NDV statistics [datafusion]
Posted by "buraksenn (via GitHub)" <gi...@apache.org>.
buraksenn commented on code in PR #20926:
URL: https://github.com/apache/datafusion/pull/20926#discussion_r3040202558
##########
datafusion/physical-plan/src/aggregates/mod.rs:
##########
@@ -1113,6 +1163,34 @@ impl AggregateExec {
}
}
+ /// Computes the estimated number of distinct groups across all grouping sets.
+ /// For each grouping set, computes `product(NDV_i + null_adj_i)` for active columns,
+ /// then sums across all sets. Returns `None` if any active column is not a direct
+ /// column reference or lacks `distinct_count` stats.
+ /// When `null_count` is absent or unknown, null_adjustment defaults to 0.
+ fn compute_group_ndv(&self, child_statistics: &Statistics) -> Option<usize> {
+ let mut total: usize = 0;
+ for group_mask in &self.group_by.groups {
+ let mut set_product: usize = 1;
+ for (j, (expr, _)) in self.group_by.expr.iter().enumerate() {
+ if group_mask[j] {
+ continue;
+ }
+ let col = expr.as_any().downcast_ref::<Column>()?;
Review Comment:
added 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] Estimate aggregate output rows using existing NDV statistics [datafusion]
Posted by "xudong963 (via GitHub)" <gi...@apache.org>.
xudong963 commented on code in PR #20926:
URL: https://github.com/apache/datafusion/pull/20926#discussion_r3043661679
##########
datafusion/physical-plan/src/aggregates/mod.rs:
##########
@@ -1113,6 +1134,82 @@ impl AggregateExec {
}
}
+ /// Estimates the output row count for grouped aggregations, combining NDV,
+ /// input row count, and TopK limit into a single [`Precision<usize>`].
+ fn estimate_num_rows(&self, child_statistics: &Statistics) -> Precision<usize> {
+ let ndv = if !self.group_by.expr.is_empty() {
+ self.compute_group_ndv(child_statistics)
+ } else {
+ None
+ };
+ let limit = self.limit_options.as_ref().map(|lo| lo.limit);
+
+ if let Some(&value) = child_statistics.num_rows.get_value() {
+ if value > 1 {
+ let mut num_rows = child_statistics.num_rows.to_inexact();
+ if let Some(ndv) = ndv {
+ num_rows = num_rows.map(|n| n.min(ndv));
+ }
+ if let Some(limit) = limit {
+ num_rows = num_rows.map(|n| n.min(limit));
+ }
+ num_rows
+ } else if value == 0 {
+ child_statistics.num_rows
+ } else {
+ let grouping_set_num = self.group_by.groups.len();
+ child_statistics.num_rows.map(|x| x * grouping_set_num)
Review Comment:
Looks like it's better to consider limit here?
--
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] Estimate aggregate output rows using existing NDV statistics [datafusion]
Posted by "asolimando (via GitHub)" <gi...@apache.org>.
asolimando commented on PR #20926:
URL: https://github.com/apache/datafusion/pull/20926#issuecomment-4197819270
> @asolimando Thanks for the detailed and thoughtful reply — really helpful and gave me a lot to think about.
>
> > If a better cardinality estimation (closer to real numbers we can observe at runtime) translates in worse plans, I think it's the cost-model that should be refined. For this reason I find #20292 very compelling for statistics, as it compares estimates vs real quantities, without the noise that the cost model could be adding.
>
> That makes sense to me. In particular, I agree with the distinction that cardinality estimation is more portable across systems, while the cost model is much more tied to a specific implementation.
>
> I’ve gone through some of those materials. This might sound a bit pessimistic, but I feel that cardinality estimation is fundamentally hard to get right, and making it mostly correct requires significant engineering effort (which may be challenging for DataFusion at the moment).
>
> My concern is mostly about how we get started in practice. Even if the long-term goal is to improve CE on its own merits, it seems useful to begin with something simple, and maybe it would require co-designs to avoid bad plans, that can be simpler than a accurate cost model.
>
> For example, in this PR’s aggregation estimation, there are at least two simple directions for estimating the output of `SELECT * FROM t GROUP BY a, b`:
>
> * Assume independence between a and b, i.e. `NDV(a) * NDV(b)`, which may overestimate.
> * Assume strong dependency, i.e. `max(NDV(a), NDV(b))`.
>
> Some system might always choose one direction, and use external mechanisms to fix bad plans. We might also want to make consistent assumptions to start, and see how to evolve overtime.
>
> Using a single reference system, or just simple heuristics are all good I think, just want to ensure we're not mixing ideas from multiple systems too early.
Thanks a lot for sharing your POV @2010YOUY01, I find it quite pragmatic and I agree on keeping it simple at the beginning and refine later when we will have better ways to evaluate the practical impact of choices around CE.
Hopefully future work on making statistics propagation overridable will also relieve some pressure as downstream systems will always be free to change how they see fit.
--
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] Estimate aggregate output rows using existing NDV statistics [datafusion]
Posted by "asolimando (via GitHub)" <gi...@apache.org>.
asolimando commented on PR #20926:
URL: https://github.com/apache/datafusion/pull/20926#issuecomment-4073101595
> Thanks @asolimando for detailed review. I thought this one was more polished after looking at Trino and Spark in detail but I've missed some important points. I'll carefully apply your reviews and adjust implementation early tomorrow
No worries, I think the PR already brings a considerable improvement over the existing, and that the final outcome will be a more precise estimation than Trino and Spark. I am bit busy today and tomorrow so don't rush unless it's good for your schedule. I will make sure to review changes by the end of the week.
--
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] Estimate aggregate output rows using existing NDV statistics [datafusion]
Posted by "buraksenn (via GitHub)" <gi...@apache.org>.
buraksenn commented on code in PR #20926:
URL: https://github.com/apache/datafusion/pull/20926#discussion_r3046674543
##########
datafusion/physical-plan/src/aggregates/mod.rs:
##########
@@ -1113,6 +1134,82 @@ impl AggregateExec {
}
}
+ /// Estimates the output row count for grouped aggregations, combining NDV,
+ /// input row count, and TopK limit into a single [`Precision<usize>`].
+ fn estimate_num_rows(&self, child_statistics: &Statistics) -> Precision<usize> {
+ let ndv = if !self.group_by.expr.is_empty() {
+ self.compute_group_ndv(child_statistics)
+ } else {
+ None
+ };
+ let limit = self.limit_options.as_ref().map(|lo| lo.limit);
+
+ if let Some(&value) = child_statistics.num_rows.get_value() {
+ if value > 1 {
+ let mut num_rows = child_statistics.num_rows.to_inexact();
+ if let Some(ndv) = ndv {
+ num_rows = num_rows.map(|n| n.min(ndv));
+ }
+ if let Some(limit) = limit {
+ num_rows = num_rows.map(|n| n.min(limit));
+ }
+ num_rows
+ } else if value == 0 {
+ child_statistics.num_rows
+ } else {
+ let grouping_set_num = self.group_by.groups.len();
+ child_statistics.num_rows.map(|x| x * grouping_set_num)
Review Comment:
I've missed that and applied it now thanks @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] Estimate aggregate output rows using existing NDV statistics [datafusion]
Posted by "asolimando (via GitHub)" <gi...@apache.org>.
asolimando commented on PR #20926:
URL: https://github.com/apache/datafusion/pull/20926#issuecomment-4224947466
@xudong963, @2010YOUY01: thanks again for your reviews and help, here and elsewhere!
For this PR specifically, is there anything missing before merging?
--
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] Estimate aggregate output rows using existing NDV statistics [datafusion]
Posted by "asolimando (via GitHub)" <gi...@apache.org>.
asolimando commented on PR #20926:
URL: https://github.com/apache/datafusion/pull/20926#issuecomment-4183702274
TLDR: I agree picking a reference system is reasonable. However, research ([Leis et al. 2015](http://www.vldb.org/pvldb/vol9/p204-leis.pdf), [Leis et al. 2018](https://doi.org/10.1007/s00778-017-0480-7), [Leis et al. 2025](http://www.vldb.org/pvldb/vol18/p5531-viktor.pdf)) shows CBO quality is dominated by cardinality estimation, not the cost model. Cardinality is an "absolute" quantity (the same plan produces the same row counts in any system), so it's portable across systems and worth improving beyond any single reference. Cost model and CBO, however, are more tied to the specific internals of any given database, and therefore less portable. As a framework, DataFusion can't assume one cost model fits all downstream use cases, so we should invest in sensible defaults for cardinality estimation while making the full statistics and cost model stack pluggable (https://github.com/apache/datafusion/issues/21120 and similar issues to come).
Apologies for the wall of text @2010YOUY01, but you raise a very deep and interesting point below, worth a detailed reply:
> ### Regarding reference system approach
> If we follow a reference-system approach, I suggest going a bit further: identify the best-performing system and try to port its related components more holistically. It’s possible that one system works significantly better than another, and that its cardinality estimation is co-designed with other components like the cost model. Taking ideas from multiple reference systems may not yield good results.
>
> The metrics for deciding reference system IMO is:
>
> 1. Simple to implement
> 2. With comprehensive documentations, explainable
> 3. Overall improvement on benchmarks like tpcds/join-order-benchmark
>
> DuckDB seems to perform well — they have a thesis (https://blobs.duckdb.org/papers/tom-ebergen-msc-thesis-join-order-optimization-with-almost-no-statistics.pdf) describing their approach, and they’ve reported benchmark results somewhere. Perhaps we can compare it with other reference systems like Spark/Trino similarly, and then pick one reference system and stick to it.
I agree it's reasonable to take a single system as reference, and that the cost model might be co-designed with its cardinality estimation to some extent. That said, I'd note a few things:
### join-order-benchmark (JOB)
Since you mentioned the JOB, which was proposed by Leis et al. in ["How Good Are Query Optimizers, Really?" (PVLDB 2015)](http://www.vldb.org/pvldb/vol9/p204-leis.pdf) exactly to go beyond TPC-DS/TPC-H and other common DB benchmarks, which aren't stressing join order algorithms enough for realistic cases.
I am a big fan of JOB, and I wish DB systems would incorporate it more systematically similarly to TPC-* benchmarks. This said, JOB has limitations too, it doesn't feature group by queries (https://github.com/gregrahn/join-order-benchmark). It therefore doesn't cover all interesting aspects of statistics/CBO, even if join-ordering is a prominent problem.
Another important conclusion from those papers is that the quality of CBO is largely dominated by that of cardinality estimation, less by the cost model. So getting cardinality right matters more than the specific cost formula consuming it.
Cardinality estimation is a proxy for a "real" and measurable quantity, the real cardinality you will observe at runtime, and it's therefore an "absolute" quantity that doesn't depend on the system: if the same exact plan was executed by any two databases, they would have the same exact number of rows at each operator.
The cost-model is, however, closer to the single implementations, as it is used to rank plans, and be able to select the "cheapest", so evaluating the cost model and CBO, is best done by judging the plan selection effect on existing benchmarks (but at the moment we don’t have lots of CBO decisions, so you won’t see effects on runtime for those benchmarks, for now).
This supports my feeling that we should strive to improve over the reference database when it makes sense.
If a better cardinality estimation (closer to real numbers we can observe at runtime) translates in worse plans, I think it's the cost-model that should be refined. For this reason I find https://github.com/apache/datafusion/pull/20292 very compelling for statistics, as it compares estimates vs real quantities, without the noise that the cost model could be adding.
### On DuckDB specifically
The thesis you linked is an interesting read, but it covers only join ordering, it won't help us much with statistics, but it will come handy for CBO (and it's an example of CBO without many detailed statistics, which is a relevant use-case for some lakehouse scenarios).
### DataFusion is a framework:
Unlike a single database like DuckDB, Trino, or Spark, DataFusion serves diverse downstream use cases where the effectiveness of statistics and CBO can vary significantly. A single DB can take more radical decisions, DataFusion should provide a sensible default but invest heavily in overridability and customization.
That's why, alongside improving the default implementation (like this PR), I'm proposing pluggable statistics machinery (like https://github.com/apache/datafusion/issues/21120 for expressions, but more to come for operator-level stats and CBO), to allow downstream systems to customize statistics, as they see fit.
--
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] Estimate aggregate output rows using existing NDV statistics [datafusion]
Posted by "asolimando (via GitHub)" <gi...@apache.org>.
asolimando commented on PR #20926:
URL: https://github.com/apache/datafusion/pull/20926#issuecomment-4183577716
> Thanks for sharing the context! happy to take a look.
Thanks @2010YOUY01 for the thoughtful feedback, let me reply in-line to be sure not to miss anything!
I will only cover in this first reply comments strictly related to this PR, the rest is very important but it's a broader CBO/stats topic, I will make a second reply just for that.
> ### Questions on this PR
> I don't have prior experience on CBO, and I'm still working through the relevant material. Here are several questions:
>
> * why we assume independence between grouping keys — is this just a simplifying heuristic, or do reference systems intentionally do this because overestimating NDV leads to better end-to-end performance (e.g., for join reordering)?
This is a standard simplifying heuristic in this space, used by both Spark and Trino. In practice, in presence of correlated columns, product would be an overestimate.
Note that we cap NDV to the number of rows to protect from severe overestimation. DuckDB defaults to number of rows directly for aggregations ([code](https://github.com/duckdb/duckdb/blob/2bdd7b54b18a99a31f4bdba240cfaede0b7fbee4/src/optimizer/statistics/operator/propagate_aggregate.cpp#L390-L391)), which is an even stronger assumption (one group per input row, only true when the grouping expressions involve a primary key or unique id).
So we have a more precise estimation on average, and we are no worse than them for the worst case thanks to NDV capping.
Keeping a main reference makes sense and would probably make our life easier, but I think it's pragmatic to improve further when there is a good opportunity and we understand the implications, like in this case.
> * why we explicitly account for nulls — this adds implementation complexity, but seems to have only a minor impact
I think the few one-liners handling nulls (`.max(1)` and adding a `+1` when nulls are present) aren't adding much complexity, considering that nulls are pretty common to have.
Cardinality estimation errors compound quickly, I think it's reasonable to trade some little more complexity for higher precision.
In general, accounting for nulls in aggregations is pretty common, you can refer to the Spark and Trino references we added as in code comments, but most other systems do handle nulls similarly.
--
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