Add skipped_aggregation_rows metric to aggregate operator

[alamb]

Which issue does this PR close?

• Closes Metrics for when partial aggregation mode is hit #11815
• Part of Improve Memory usage + performance with large numbers of groups / High Cardinality Aggregates #6937
• Follow on to Skipping partial aggregation when it is not helping for high cardinality aggregates #11627

Rationale for this change

#11627 from @korowa adds a "partial aggregation skipping" mode to the hash aggregate exec that switches aggregate behavior dynamically at runtime,

It would be very nice to know if the path is being executed or not, and the way to do this in DataFusion is metrics.

What changes are included in this PR?

Add a "skipped_aggregation_rows" counter which records the number of rows

Are these changes tested?

I tested them manually

For example, this line shows skipped_aggregation_rows=98293561:

|                   |               AggregateExec: mode=Partial, gby=[WatchID@0 as WatchID, ClientIP@1 as ClientIP], aggr=[count(*)], metrics=[output_rows=99997497, elapsed_compute=190.246833ms, skipped_aggregation_rows=98293561]                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                      |

Here is the entire output of testing with a query:

$ ./datafusion-cli-skip-partial-metrics -c "EXPLAIN ANALYZE SELECT \"WatchID\", \"ClientIP\", COUNT(*) AS c FROM 'hits.parquet' GROUP BY \"WatchID\", \"ClientIP\" ORDER BY c DESC LIMIT 10;"
DataFusion CLI v40.0.0
+-------------------+--------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+
| plan_type         | plan                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                 |
+-------------------+--------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+
| Plan with Metrics | GlobalLimitExec: skip=0, fetch=10, metrics=[output_rows=10, elapsed_compute=14.916µs]                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                |
|                   |   SortPreservingMergeExec: [c@2 DESC], fetch=10, metrics=[output_rows=10, elapsed_compute=3.667µs]                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                   |
|                   |     SortExec: TopK(fetch=10), expr=[c@2 DESC], preserve_partitioning=[true], metrics=[output_rows=160, elapsed_compute=764.776988ms, row_replacements=164]                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                           |
|                   |       ProjectionExec: expr=[WatchID@0 as WatchID, ClientIP@1 as ClientIP, count(*)@2 as c], metrics=[output_rows=99997493, elapsed_compute=1.960225ms]                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                               |
|                   |         AggregateExec: mode=FinalPartitioned, gby=[WatchID@0 as WatchID, ClientIP@1 as ClientIP], aggr=[count(*)], metrics=[output_rows=99997493, elapsed_compute=17.934170583s]                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                     |
|                   |           CoalesceBatchesExec: target_batch_size=8192, metrics=[output_rows=99997497, elapsed_compute=598.045136ms]                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                  |
|                   |             RepartitionExec: partitioning=Hash([WatchID@0, ClientIP@1], 16), input_partitions=16, metrics=[repart_time=1.018381539s, send_time=8.695161456s, fetch_time=1.684569996s]                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                |
|    here-->        |               AggregateExec: mode=Partial, gby=[WatchID@0 as WatchID, ClientIP@1 as ClientIP], aggr=[count(*)], metrics=[output_rows=99997497, elapsed_compute=190.246833ms, skipped_aggregation_rows=98293561]                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                      |
|                   |                 ParquetExec: file_groups={16 groups: [[Users/andrewlamb/Downloads/hits.parquet:0..923748528], [Users/andrewlamb/Downloads/hits.parquet:923748528..1847497056], [Users/andrewlamb/Downloads/hits.parquet:1847497056..2771245584], [Users/andrewlamb/Downloads/hits.parquet:2771245584..3694994112], [Users/andrewlamb/Downloads/hits.parquet:3694994112..4618742640], ...]}, projection=[WatchID, ClientIP], metrics=[output_rows=99997497, elapsed_compute=16ns, page_index_rows_filtered=0, row_groups_matched_statistics=0, row_groups_matched_bloom_filter=0, file_scan_errors=0, row_groups_pruned_bloom_filter=0, predicate_evaluation_errors=0, bytes_scanned=1050612937, pushdown_rows_filtered=0, file_open_errors=0, num_predicate_creation_errors=0, row_groups_pruned_statistics=0, pushdown_eval_time=32ns, time_elapsed_processing=978.395665ms, page_index_eval_time=32ns, time_elapsed_scanning_until_data=68.642586ms, time_elapsed_scanning_total=16.706024614s, time_elapsed_opening=380.053666ms] |
|                   |                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                      |
+-------------------+--------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+
1 row(s) fetched.
Elapsed 1.485 seconds.

Are there any user-facing changes?

Another metric in a plan if partial aggregation is skipped

[alamb]

This is the actual call that records the metrics . The rest of the PR is comments and plumbing

[korowa]

Agree that it should be better not to add this counter into all group bys (as another baseline metric) by default 👍

[alamb]

FYI @Dandandan would you have time to review this PR (to help performance profiling of the group by aggregation skipping code)?

[andygrove]

LGTM. Thanks @alamb

[alamb]

Thanks for the review @andygrove and @korowa