Support rate family aggregation functions in table model

[CoollZzz]

Description

Related issue: #17976

This PR adds four built-in aggregation functions to the IoTDB table model: rate(), increase(), irate(), and delta(). Their calculation semantics are based on the corresponding Prometheus range-vector functions, while window boundaries follow IoTDB's [window_start, window_end) convention.

Function semantics

Function	Signature	Description
rate	rate(value, time, window_start, window_end)	Calculates the extrapolated average per-second growth rate of a counter and compensates for counter resets.
increase	increase(value, time, window_start, window_end)	Calculates the extrapolated total increase of a counter and compensates for counter resets.
irate	irate(value, time)	Calculates the instantaneous per-second counter rate using the last two valid samples.
delta	delta(value, time, window_start, window_end)	Calculates the extrapolated difference of a gauge without counter-reset compensation.

The supported value types are INT32, INT64, FLOAT, and DOUBLE. Time and window arguments accept TIMESTAMP and INT64, with INT64 interpreted using the current timestamp_precision. All four functions return DOUBLE.

rate(), increase(), and delta() use all valid samples in the aggregation group and extrapolate the result to the window boundaries. Extrapolation uses the average sampling interval, a 1.1 threshold for distant boundaries, and counter zero-point protection where applicable. irate() only uses the last two valid samples and does not perform boundary extrapolation.

Accumulator and planner design

Both TableAccumulator and GroupedAccumulator execution paths are supported.

Each function has its own abstract accumulator class for shared function-specific validation, window handling, and result calculation. Ordered and naive accumulators are implemented as separate concrete classes because their state structures and lifecycle behavior differ significantly.

• Ordered accumulators are selected only for the SINGLE aggregation step when the physical planner proves that the function's time argument is ascending within every SQL aggregation group. They consume samples incrementally without buffering and sorting all input rows.
• Naive accumulators buffer samples and sort them by time before evaluation. They support unordered input and all multi-stage aggregation steps.
• Multi-stage aggregation uses a versioned BLOB intermediate state containing the complete sample state and window boundaries.
• Naive table and grouped accumulators report retained buffer memory through the existing memory reservation framework.
• The four functions are not pushed into aggregation table scans because their explicit time argument may differ from the physical table time column and their calculations require raw samples.

The ordered-input property is propagated through AggregationNode serialization so that distributed execution selects the same accumulator implementation as the coordinator plan.

Validation and edge cases

The implementation handles the following cases:

• NULL values are ignored.
• Fewer than two valid samples return NULL.
• Counter resets are compensated for rate(), increase(), and irate().
• Negative values are rejected for counter functions but are supported by delta().
• Non-finite values, including NaN and positive or negative infinity, are rejected.
• Required time or window arguments cannot be NULL for a valid sample.
• Window boundaries must be consistent within an aggregation group and satisfy window_start < window_end.
• Every valid sample must satisfy window_start <= time < window_end.
• Duplicate timestamps within the same aggregation group are rejected.
• Non-finite intermediate or extrapolated results are rejected.

Tests

Integration tests were added to IoTDBTableAggregationIT for:

• Normal global, grouped, ordered, unordered, distributed, and windowed aggregation.
• Counter resets, multiple resets, zero-point protection, extrapolation thresholds, null values, missing samples, and gap-fill scenarios.
• All supported numeric and time argument types.
• Invalid argument counts and types, non-finite and negative counter values, invalid or inconsistent windows, out-of-window samples, null time arguments, and duplicate timestamps.

The four target test methods were verified with both the TableSimpleIT and TableClusterIT profiles. The full reactor test compilation also passed for both the default and Chinese locales.

---

This PR has:

• been self-reviewed.
• added integration tests.
• been tested in a test IoTDB cluster.

---

Key changed/added classes (or packages if there are too many classes) in this PR

• org.apache.iotdb.calc.execution.operator.source.relational.aggregation.rate
• org.apache.iotdb.calc.execution.operator.source.relational.aggregation.grouped.rate
• AccumulatorFactory
• TableMetadataImpl
• TableDistributedPlanGenerator
• AggregationNode
• PushAggregationIntoTableScan
• IoTDBTableAggregationIT

[JackieTien97]

Thanks for adding the table-model rate-family aggregations. I reviewed the implementation against the requirement and technical design and found blocking correctness and memory-accounting issues in the inline comments.

Please also complete the following coverage and delivery items:

• Run behavioral tests under ms, us, and ns timestamp_precision, including exact-threshold and one-tick-below cases.
• Link the companion user-manual change for the table SQL functions/operators if documentation is maintained in a separate repository.
• Add focused unit coverage for the extrapolation math, intermediate-state codec, planner ordered/naive selection, and AggregationNode serialization.

Requesting changes because the threshold comparison can return incorrect query results, and the intermediate-state path can exceed the query memory budget.

[JackieTien97]

[P1] Compare the extrapolation threshold in integer tick space

The contract requires using half the average interval when gap >= 1.1 * averageInterval, but converting each interval independently to double seconds can flip this boundary. At millisecond precision, for window [0, 31) with samples (11, 100) and (21, 110), the left gap is exactly 1.1 * 10 ms. The expected increase/delta is 25 and rate is about 806.4516; here threshold becomes 0.011000000000000001 while the gap is 0.011, so the branch is skipped and the implementation returns 31 and 1000.

Please compare exact ticks before converting to seconds, for example with BigInteger: gapTicks * (sampleCount - 1) * 10 >= sampledTicks * 11. Add regression cases for both boundaries under ms/us/ns precision, covering equality and one tick below the threshold.

[JackieTien97]

[P1] Reserve or bound the serialized rate state before allocating it

This materializes every sample in the group into one additional header + 16 * n byte array while the original TimeValueBuffer is still live. The allocation is not reserved through the query memory manager, a single Binary is not bounded by maxTsBlockSizeInBytes, and AggregationOperator.calculateMaxReturnSize() still advertises only the fixed max TsBlock size. A valid partial aggregation can therefore fit its registered accumulator reservation and then exceed that budget—potentially with a single state approaching 2 GB—while serializing. Decode also retains the input Binary while creating a temporary buffer and merging it into the destination.

Please either chunk/bound the intermediate state or reserve the serialized output and decode-temporary memory before allocation.

[JackieTien97]

[P2] Avoid charging grouped rate state twice

This accumulator reserves its full estimated size through the fragment-level memory manager, but HashAggregationOperator and StreamingHashAggregationOperator reserve the same state again through InMemoryHashAggregationBuilder#getEstimatedSize(), which already includes every GroupedAggregator. Both paths resolve to the same FragmentInstanceContext manager, so allocations in all eight grouped rate-family accumulators are charged roughly twice and valid grouped queries can fail with MemoryNotEnoughException while the actual state still fits.

Please make either the grouped accumulator or the enclosing hash operator the single owner of this reservation.

[JackieTien97]

[P3] Preserve the current/previous timestamp order in the Chinese message

The callers format this template with (functionName, currentTime, previousTime), matching the English wording got current after previous. This Chinese wording reads the placeholders in the opposite order: when time moves from 2 back to 1, it reports 在 1 之后得到了 2. Please either rewrite it as 但得到了 %d，前一个值为 %d or use positional placeholders that preserve the supplied argument order.

[JackieTien97]

[P2] Exercise an actual multi-stage intermediate merge here

Both fixture batches use the same scenario=distributed TAG. Data-partition inheritance keeps all time slots for the same SeriesSlot in the same DataRegion, so spanning two files and time partitions does not demonstrate the cluster partial/intermediate/final merge claimed by this comment.

Please use multiple tag/device values placed in different DataRegions while aggregating them into the same SQL group, and assert the placement or physical plan so this test reliably covers merging multiple intermediate states.

[JackieTien97]

[P2] Cover an overlapping HOP window

With SLIDE => 5ms and SIZE => 5ms, no input row belongs to multiple windows, so this exercises the same window shape as TUMBLE rather than the overlapping HOP scenario required by the specification. Please add a case with SLIDE < SIZE and verify the rate-family results for every overlapping window.