Estimate aggregate output rows using existing NDV statistics #20926
Estimate aggregate output rows using existing NDV statistics #20926buraksenn wants to merge 3 commits intoapache:mainfrom
Conversation
asolimando
left a comment
asolimando
left a comment
There was a problem hiding this comment.
Thanks @buraksenn for working on this, I have left a few comments, hoping this helps!
| 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, |
There was a problem hiding this comment.
Nit: it's a reasonable default but I'd add this to the comment of the function explicitly
| 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); |
There was a problem hiding this 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 is bailing out for (multiple) grouping sets, Spark 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;
| child_statistics.num_rows.map(|x| x * grouping_set_num) | ||
| } | ||
| } else if let Some(limit_opts) = &self.limit_options { | ||
| Precision::Inexact(limit_opts.limit) |
There was a problem hiding this comment.
Here we might have ndv even if num_rows is unset, maybe we can return Inexact(min(ndv, limit))?
| 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()?; |
There was a problem hiding this 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.
|
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. |
2 participants
Which issue does this PR close?
Part of #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.
What changes are included in this PR?
Are these changes tested?
Yes existing and new tests that cover different scenarios and edge cases
Are there any user-facing changes?
No