feat: Improve metrics for aggregate streams.

[EmilyMatt]

Which issue does this PR close?

• Closes Improve metrics for Aggregate streams #18323 .

Rationale for this change

Adds more detailed metrics, so it is easier to identify which part of the aggregate streams are actually slow.

What changes are included in this PR?

Added a metrics struct, and used it in the functions common to the aggregate streams.

Are these changes tested?

Yes, added some tests to verify the metrics are actually updated and can be retrieved.

I've also ran the groupby benchmarks to ensure we don't create timers in a way that could impact performance, and it seems ok, all the changes are within what I'd expect as std variation on a local machine.

Comparing main and agg-metrics
--------------------
Benchmark h2o.json
--------------------
┏━━━━━━━━━━━━━━┳━━━━━━━━━━━━┳━━━━━━━━━━━━━┳━━━━━━━━━━━━━━━┓
┃ Query        ┃       main ┃ agg-metrics ┃        Change ┃
┡━━━━━━━━━━━━━━╇━━━━━━━━━━━━╇━━━━━━━━━━━━━╇━━━━━━━━━━━━━━━┩
│ QQuery 1     │ 1252.42 ms │  1196.62 ms │     no change │
│ QQuery 2     │ 3976.62 ms │  3392.89 ms │ +1.17x faster │
│ QQuery 3     │ 3448.29 ms │  2918.47 ms │ +1.18x faster │
│ QQuery 4     │ 1909.15 ms │  1632.98 ms │ +1.17x faster │
│ QQuery 5     │ 3056.36 ms │  2831.82 ms │ +1.08x faster │
│ QQuery 6     │ 2663.13 ms │  2594.64 ms │     no change │
│ QQuery 7     │ 2802.28 ms │  2592.43 ms │ +1.08x faster │
│ QQuery 8     │ 4489.29 ms │  4199.00 ms │ +1.07x faster │
│ QQuery 9     │ 7001.75 ms │  6622.98 ms │ +1.06x faster │
│ QQuery 10    │ 4725.80 ms │  4619.37 ms │     no change │
└──────────────┴────────────┴─────────────┴───────────────┘
┏━━━━━━━━━━━━━━━━━━━━━━━━━━━━┳━━━━━━━━━━━━┓
┃ Benchmark Summary          ┃            ┃
┡━━━━━━━━━━━━━━━━━━━━━━━━━━━━╇━━━━━━━━━━━━┩
│ Total Time (main)          │ 35325.09ms │
│ Total Time (agg-metrics)   │ 32601.19ms │
│ Average Time (main)        │  3532.51ms │
│ Average Time (agg-metrics) │  3260.12ms │
│ Queries Faster             │          7 │
│ Queries Slower             │          0 │
│ Queries with No Change     │          3 │
│ Queries with Failure       │          0 │
└────────────────────────────┴────────────┘

Are there any user-facing changes?

Nothing that is direct to the user, additional metrics will now be available, but no breaking changes.

[rluvaton]

this is a breaking change as you changed the visibility and the arguments.

you should create a new function and leave this function as is

[EmilyMatt]

Yes, that is mentioned in the PR

[rluvaton]

I don't think we should change that function even when it is a utility

[EmilyMatt]

I think this is a sure way to end up with lots of dead code because "someone surely uses it", instead, it was simply a mistake to have such a generic utility in this file made public, and the earlier it is fixed the less people are affected.

[rluvaton]

this was not a mistake but an inform decision

• chore: Make GroupValues and APIs on PhysicalGroupBy aggregation APIs public #16733

[EmilyMatt]

Then this function should have been moved somewhere else, there are multiple copies of the code that does exactly that all over the project, keeping this specific instance pub is super random, and slightly disruptive, will work around it to not modify the visibility in this PR's scope and will open an issue to change that

[rluvaton]

Observation: the aggregation time also include finding the group indices which I don't think it

however a possible solution is to move inside the loop but I'm not a fan of it as it can be a hot loop

[EmilyMatt]

Yes that was a conscious choice to avoid running this once per group, because current_group_indices is actually a scratch space, it can't organically be separated into two loops(although creating one scratch space per group_by is also valid)

[rluvaton]

the loop also contain building and searching the hash table self.group_values.intern which can be expensive (e.g. when GroupValuesRows is used)

[EmilyMatt]

Added a metric for it, I think this is a mistake tho

[rluvaton]

Can you also add a timer for building and searching (single timer) the the hash table (group_values.intern)

or if you prefer we can leave that for later

[EmilyMatt]

I think this will become a performance hit, I'd rather keep metrics concrete to avoid syscalls

[rluvaton]

Can you add few subset timers from this:

1. building aggregate expressions (how much time calling all accumulators update_batch/merge_batch)
2. emitting aggregate expressions (how much time calling all accumulators emit/state)
3. Emitting grouping columns (self.group_values.emit in row_hash is an example of that emission)

[EmilyMatt]

Same as previous comments, I think using clocks here is going to start encumbering the system

[rluvaton]

Can you please update the PR description after your changes?

[rluvaton]

LGTM, thanks @EmilyMatt !