Add metrics for Limit and Projection, and CoalesceBatches

[alamb]

Which issue does this PR close?

Next part #866 (following the same model as #960).

Rationale for this change

We want basic understanding of where a plan's time is spent and in what operators. See #866 for more details

What changes are included in this PR?

1. Instrument ProjectionExec, GlobalLimitExec, LocalLimitExec and CoalesceBatchesExec using the API from Add BaselineMetrics, Timestamp metrics, add for CoalescePartitionsExec, rename output_time -> elapsed_compute #909
2. Tests for same

Are there any user-facing changes?

More fields in EXPLAIN ANALYZE are now filled out

Example of how explain analyze is looking (dense but packed with good info)

running query: EXPLAIN ANALYZE selectcount(*) from (SELECTcount(*), c1 FROM aggregate_test_100 WHERE c13 !='C2GT5KVyOPZpgKVl110TyZO0NcJ434'GROUP BY c1 ORDER BY c1) LIMIT1
Query Output:
+-------------------+-------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+
| plan_type | plan |
+-------------------+-------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+
| Plan with Metrics | GlobalLimitExec: limit=1, metrics=[output_rows=1, elapsed_compute=182ns] |
| | ProjectionExec: expr=[COUNT(UInt8(1))@0asCOUNT(UInt8(1))], metrics=[output_rows=1, elapsed_compute=7.556µs] |
| | HashAggregateExec: mode=Final, gby=[], aggr=[COUNT(UInt8(1))], metrics=[output_rows=1, elapsed_compute=48.283µs] |
| | CoalescePartitionsExec, metrics=[output_rows=3, elapsed_compute=NOT RECORDED] |
| | HashAggregateExec: mode=Partial, gby=[], aggr=[COUNT(UInt8(1))], metrics=[output_rows=3, elapsed_compute=101.291µs] |
| | RepartitionExec: partitioning=RoundRobinBatch(3), metrics=[fetch_time{inputPartition=0}=10.721068ms, send_time{inputPartition=0}=3.93µs, repart_time{inputPartition=0}=NOT RECORDED] |
| | SortExec: [c1@0ASC], metrics=[output_rows=5, elapsed_compute=184.753µs] |
| | CoalescePartitionsExec, metrics=[output_rows=5, elapsed_compute=NOT RECORDED] |
| | HashAggregateExec: mode=FinalPartitioned, gby=[c1@0as c1], aggr=[COUNT(UInt8(1))], metrics=[output_rows=5, elapsed_compute=299.555µs] |
| | CoalesceBatchesExec: target_batch_size=4096, metrics=[output_rows=5, elapsed_compute=73.812µs] |
| | RepartitionExec: partitioning=Hash([Column { name: "c1", index: 0 }], 3), metrics=[send_time{inputPartition=0}=NOT RECORDED, fetch_time{inputPartition=0}=28.696948ms, repart_time{inputPartition=0}=201.886µs] |
| | HashAggregateExec: mode=Partial, gby=[c1@0as c1], aggr=[COUNT(UInt8(1))], metrics=[output_rows=5, elapsed_compute=560.718µs] |
| | CoalesceBatchesExec: target_batch_size=4096, metrics=[output_rows=99, elapsed_compute=306.864µs] |
| | FilterExec: c13@1!= C2GT5KVyOPZpgKVl110TyZO0NcJ434, metrics=[output_rows=99, elapsed_compute=264.367µs] |
| | RepartitionExec: partitioning=RoundRobinBatch(3), metrics=[repart_time{inputPartition=0}=NOT RECORDED, send_time{inputPartition=0}=5.482µs, fetch_time{inputPartition=0}=6.933244ms] |
| | CsvExec: source=Path(/Users/alamb/Software/arrow/testing/data/csv/aggregate_test_100.csv: [/Users/alamb/Software/arrow/testing/data/csv/aggregate_test_100.csv]), has_header=true, metrics=[] |
+-------------------+-------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+

[Dandandan]

Great addition as always 👍

[alamb]

BTW here are some cool screen shots of using this data in IOx to visualize DataFusion execution plan progression: https://github.com/influxdata/influxdb_iox/pull/2529 if anyone is interested

[Dandandan]

Amazing 😎

[houqp]

really cool jaeger indeed!