New execution model for rates

[dnhatn]

Our current execution path for rate aggregation, particularly for calculating counter resets, requires data in each bucket to be strictly ordered by timestamp. This necessitates a specialized execution path for time-series data, which I believe is unnecessarily complex.

I've explored an alternative model: instead of pre-sorting data in the source operator, we buffer data in each bucket and perform a merge-sort just before emitting the output. This would eliminate the need for specialized time-series code and allow us to leverage existing ES|QL optimizations.

The main downside is the memory usage for buffering rate points. Each data point requires about 16 bytes; typical queries over a few million points would use less than 100MB, but worst-case scenarios could consume up to 32GB, potentially causing circuit breaking errors.

We can mitigate this with the following enhancements:

1. Execute segments in descending max_timestamp order: By processing segments this way, the source operator can provide a "high-water mark" (the maximum timestamp that may appear in the current or subsequent segments). This allows the rate aggregator to safely flush any buffered data that is more recent than this mark, keeping the buffer size minimal or avoid buffering data points.

2. Dynamically split shards by time interval: For large time ranges with interleaved data, we can partition execution into smaller time intervals based on min and max timestamps. This limits buffer size and improves parallelism.

This PR is the first step, it cut over from the current to the new rate with buffer. This new rate still delegates to the old rate after merging the buffer.

I benchmarked this change for the below query, the executiton time reduced from 405ms -> 270ms.

TS my* 
| WHERE `metrics.system.cpu.time` IS NOT NULL AND @timestamp >= \"2025-07-25T14:55:59.000Z\" AND @timestamp <= \"2025-07-25T16:25:59.000Z\"
| STATS AVG(RATE(`metrics.system.cpu.time`)) BY host.name, BUCKET(@timestamp, 1h) 
| LIMIT 10000

I expect more to come with high-water mark.

Relates #134324

[dnhatn]

These new rate classes are the main change; the rest removes the specialized execution path for rates.

[dnhatn]

This was moved because Javadoc does not work with code-generated classes. This class will be removed in a follow-up.

[elasticsearchmachine]

Pinging @elastic/es-storage-engine (Team:StorageEngine)

[elasticsearchmachine]

Pinging @elastic/es-analytical-engine (Team:Analytics)

[kkrik-es]

Nit: move outside the method as a private class for readability.

[kkrik-es]

I'm confused with this one.. Mind adding a comment about how this works? I'd think (naively) that we only need to keep adding dv(val, prevValue) to the delta as that one detects resets.

[kkrik-es]

I think this formula allows tracking (a) the delta before resetting, (b) using 0 as the new low bound for values. If so, worth documenting in a comment.

[kkrik-es]

Nit: the definition and initialization of the priority queue can also be moved into Slice for readability.

[kkrik-es]

Consider adding a comment outlining how the buffer works, i.e. storing multiple slices in a single array and tracking the start offset of each one.

[kkrik-es]

Add a comment that timestamps appearing out of order indicate the start of a new slice.

[kkrik-es]

Nit: this loop seems to belong more to Buffer :)

It can return an array of slices to be fed then to the priority queue.

[kkrik-es]

To confirm my understanding, this runs in a single thread per shard per bucket, after we collect all the data from all segments for this bucket?

[kkrik-es]

This is really, really cool - and you get to clean up so much special logic as a bonus. Just some nits to improve readability.

The comment about in-mem buffering is legit. I wonder if we can fall back to the slow path if we estimate that the buffers will exceed e.g. 1% of available memory.

[dnhatn]

Thanks Kostas! I have addressed all of your comments.