Hi xinqi, Do you think about pushing these four functions down to TableScan and combining AggNode with TableaScan into an AggTableScanNode. If so, why do you choose to not do that?
Best regards, --------------------- Yuan Tian On Fri, Jul 24, 2026 at 10:33 AM Xinqi Zhao <[email protected]> wrote: > Hi Yuan and all, > > Thank you for the detailed review and suggestions. I have revised the > technical design accordingly. > > Related issue: > https://github.com/apache/iotdb/issues/17976 > > The main changes are summarized below. > > Separate Ordered and Naive accumulator implementations > > The Ordered and Naive modes have been split into different concrete > classes because they maintain different states and have different lifecycle > requirements. > > For each function, an abstract base class now contains the shared > validation, window semantics, and final calculation logic, while the > concrete subclasses implement the corresponding state management: > > AbstractRateTableAccumulator > > OrderedRateAccumulator > NaiveRateAccumulator > > The same structure is used for increase(), irate(), and delta(), as well > as for the GroupedAccumulator path. > > The accumulator path and the algorithm are treated as two independent > dimensions: > > The operator selects TableAccumulator or GroupedAccumulator. > The aggregation step and ordering property select Ordered or Naive. > > Ordered implementations are used only for SINGLE aggregation and keep O(1) > state per group. Naive implementations preserve all samples and support > SINGLE, PARTIAL, INTERMEDIATE, and FINAL stages. > > Explicit derivation of the ordered-input property > > The updated design defines inputOrderedByTimeAscending as a > per-aggregation physical property. It is derived in > TableDistributedPlanGenerator.visitAggregation() after the final physical > child of the AggregationNode has been established. > > The proof uses the final child’s OrderingScheme. For a particular > aggregation: > > The second argument must be a directly addressable time_col symbol. > Before time_col appears in the ordering prefix, only SQL grouping keys are > allowed. > time_col must use ascending order. > Ordering symbols after time_col do not affect the proof. > If the ordering is missing, uses descending time, contains a non-grouping > symbol before time_col, or otherwise cannot be proven, the property is > false. > Arbitrary time expressions conservatively use the Naive implementation. > > This also means that Project, Exchange, Join, Union, HOP/TUMBLE, and other > operators are handled through the ordering property of the finalized > physical child. If they do not preserve and expose a sufficient > OrderingScheme, the Ordered implementation is not selected. > > Only the following condition enables the Ordered implementation: > > step == SINGLE && inputOrderedByTimeAscending > > PARTIAL, INTERMEDIATE, and FINAL stages always use the Naive > implementation. > > The current version also disables aggregation pushdown to > AggregationTableScanNode and ExternalTsFileAggregationScanNode for these > functions. The explicit time_col may differ from the physical TIME column, > and the functions require raw samples rather than file, chunk, or page > statistics. > > Common runtime validation flow > > A stateless RateFunctionValidation utility is now shared by Ordered/Naive, > Table/Grouped, and the different aggregation stages. > > The validation order and behavior are explicitly defined: > > Rows with NULL value_col are ignored. > If value_col is non-NULL, the required time and window arguments must be > non-NULL. > NaN and Infinity are rejected. > Negative values are rejected for rate(), increase(), and irate(), while > delta() accepts negative values. > window_start must be less than window_end. > Samples must satisfy window_start <= time_col < window_end. > Window boundaries must be consistent within the same SQL group. > Duplicate timestamps are rejected. > Ordered implementations also reject an unexpected decrease in timestamps. > After filtering, fewer than two valid samples produce NULL. > > Regarding NULL handling, I retained the requirement-level contract that > only a NULL value_col causes a row to be ignored. A non-NULL value with a > NULL time or required window boundary is treated as invalid input. This > rule is now applied consistently in every execution path. > > Versioned Intermediate State protocol > > The exact BLOB formats are now specified as follows: > > rate(), increase(), and delta(): > > version:int32 > windowStart:int64 > windowEnd:int64 > sampleCount:int32 > repeated { timestamp:int64, value:double } > > irate(): > > version:int32 > sampleCount:int32 > repeated { timestamp:int64, value:double } > > The initial protocol version is 1. > > The state rules are now explicit: > > A partial state with no valid samples is represented as NULL. > A one-sample state is preserved and encoded normally. > NULL intermediate positions are skipped. > Window boundaries are checked for consistency when states are merged. > Samples from different states are merged without relying on state arrival > order. > Duplicate timestamps are checked after all states have been merged and > sorted. > The codec validates the version, header and total length, sample count, > window values, timestamps, and sample values before creating the decoded > buffer. > Serialized-size calculations use exact arithmetic and must fit the > ByteBuffer integer capacity. > > I also corrected the TimeValueBuffer.merge() issue identified in the > previous pseudocode: the original size is retained as the copy offset, and > the size is updated only once after copying. > > Incremental grouped memory accounting > > Grouped Naive accumulators no longer traverse every non-empty group after > each input batch. > > TimeValueBufferBigArray maintains buffersRetainedBytes incrementally. Only > buffers modified by the current operation, together with any ObjectBigArray > capacity change, contribute to the memory delta. > > As a result: > > Appending one sample has amortized O(1) accounting cost. > Merging a state is proportional to the number of merged samples. > Processing a batch is proportional to the positions or groups actually > visited. > There is no additional O(blockCount × groupCount) scan. > > Final sorting is performed in place using introsort on the parallel > timestamp and value arrays. Its worst-case time complexity is O(n log n), > with O(log n) auxiliary space. > > Numerical edge cases > > The numerical rules have also been clarified: > > Timestamp subtraction uses BigInteger before conversion to seconds, > avoiding long overflow while preserving exact tick differences. > Window and sample intervals are validated before applying the increase == > 0 shortcut. > Intermediate and final arithmetic results are checked for finiteness. > A non-finite durationToZero does not cause zero-point truncation and is > not itself treated as an error. > sampleCount uses int and is updated with exact overflow checks. The > serialized state is also limited by the maximum ByteBuffer size. > All input values are represented internally as double. Therefore, INT64 > values whose absolute value exceeds 2^53 follow IEEE 754 double semantics, > and unit-level differences are not guaranteed to remain exact. This > limitation is now documented explicitly. > > The revised design continues to preserve complete samples during > distributed aggregation and performs sorting, duplicate detection, > counter-reset correction, and extrapolation only after all states have been > merged. > > Best regards, > Xinqi Zhao > > > > > > 原始邮件 > 发件人：Yuan Tian <[email protected]> > 发件时间：2026年7月17日 11:57 > 收件人：dev <[email protected]> > 主题：Re: [DISCUSS] Technical design for rate(), irate(), increase(), and > delta() aggregate functions > > > Hi Xinqi, > > Thanks for sharing the technical design. I reviewed the detailed design > document and left several inline comments. Overall, reusing the existing > table-model aggregation framework and preserving complete samples during > distributed aggregation looks reasonable. Before implementation, I think > the following points should be clarified or corrected. > > 1. Accumulator organization > > The ordered and buffered modes maintain substantially different states and > follow different processing paths. I suggest reconsidering whether both > modes should live in the same concrete accumulator class. A common abstract > base class could contain shared validation and calculation logic, while the > ordered and buffered implementations could be separate subclasses. This may > > reduce mode-specific branching and make their respective invariants clearer. > > 2. Derivation of the ordered-input property > > The design says that the ordered implementation is selected when the > planner can prove that samples are ordered by time_col, but it does not yet > define how this proof is made. > > The design should specify: > > - At which physical-planning stage the property is derived. > - How per-group ordering, rather than only global ordering, is proven. > - How Project, Exchange, Join, Union, HOP/TUMBLE, and other operators > affect the property. > - Whether only a direct time_col symbol is supported; an arbitrary time > > expression should conservatively fall back to the buffered implementation. > - That the property must be false whenever ordering cannot be proven. > > Although inputOrderedByTimeAscending is not meaningful for every > aggregation function, keeping it in AggregationNode.Aggregation as a > serialized per-aggregation physical hint seems acceptable if its scope and > exact meaning are clearly documented. The derivation should happen after > the physical input is finalized rather than during logical function > analysis. > > 3. Runtime validation contract > > The design should define a common validation flow shared by > ordered/buffered, Table/Grouped, and SINGLE/PARTIAL/FINAL paths. In > particular: > > - Ignore samples whose value or time is NULL before validating valid > samples. > - Reject NaN and Infinity. > - Reject negative counter values for rate(), increase(), and irate(); > delta() may accept negative values. > - Require non-NULL window bounds and window_start < window_end. > - Require sample timestamps to be within [window_start, window_end). > - Require consistent window bounds within the same group. > - Reject duplicate timestamps. > - Return NULL when fewer than two valid samples remain. > > Without one explicitly defined common entry point and validation order, the > different accumulator paths may produce inconsistent behavior. > > 4. Intermediate-state format and merge behavior > > The exact binary format should be specified, for example: > > rate/increase/delta: > version | windowStart | windowEnd | sampleCount | samples > > irate: > version | sampleCount | samples > > The design should also define: > > > - Whether a partial state with no valid samples is NULL or has sampleCount > = 0. > - How a one-sample partial state is preserved. > - How addIntermediate() handles NULL positions. > - How inconsistent window bounds and duplicate timestamps across states > are handled. > - How unknown versions, truncated states, trailing bytes, and oversized > BLOBs are handled. > > There is also a correctness issue in the current TimeValueBuffer.merge() > pseudocode: it assigns size = mergedSize before copying and then copies > from offset size, which will cause an out-of-bounds access. The old size > should be retained as the copy offset, and size should be updated only once > after the copy. > > 5. Grouped memory accounting > > The proposed grouped implementation recalculates retained memory by > traversing every non-empty group after each input or intermediate batch. > This adds approximately O(blockCount × groupCount) work and may approach > O(n²) in unfavorable cases. > > It would be better to maintain the retained-size delta incrementally for > only the buffers changed by the current batch, including any ObjectBigArray > capacity change. > > 6. Numerical edge cases > > The extrapolation design should also address the following cases: > > - Math.subtractExact(later, earlier) may overflow for two otherwise > valid INT64 timestamps. > - The final division performed by rate() or irate() may still produce > Infinity even if the extrapolated intermediate result is finite. > - The increase == 0 shortcut should occur only after validating the > window and sample interval. > - Converting INT64 values greater than 2^53 to double can lose unit > increments; the intended semantics should be documented. > - The upper bound implied by using an int for sampleCount should be > defined. > > These points do not change the overall direction of the design, but they > affect correctness, consistency, and performance. I suggest making them > explicit in the technical design before implementation begins. > > Best regards, > --------------------- > > Yuan Tian > > On Fri, Jul 17, 2026 at 11:02 AM Xinqi Zhao <[email protected]> wrote: > > > Hi IoTDB community, > > > > > Following the previous requirements discussion, I have prepared an initial > > technical design for adding the Prometheus-like rate(), irate(), > > increase(), and delta() aggregate functions to the IoTDB table model. > > > > Related issue: > > https://github.com/apache/iotdb/issues/17976 > > > > The main design is summarized below. > > > > Integration with the existing aggregation framework > > > > The four functions will be implemented entirely within the existing > > table-model aggregation framework. No new SQL execution layer or > > third-party dependency will be introduced. > > > > Both execution paths will be supported: > > > > TableAccumulator for global aggregation and streaming aggregation when > > groups are fully ordered. > > > > GroupedAccumulator for HashAggregationOperator and > > StreamingHashAggregationOperator. > > > > Each function will provide both TableAccumulator and GroupedAccumulator > > implementations. > > > > The signatures are: > > > > rate(value_col, time_col, window_start, window_end) > > increase(value_col, time_col, window_start, window_end) > > irate(value_col, time_col) > > delta(value_col, time_col, window_start, window_end) > > > > The value column supports INT32, INT64, FLOAT, and DOUBLE. Values are > > > converted to double internally, and all four functions return DOUBLE. Time > > arguments support TIMESTAMP and INT64. > > > > Ordered and unordered implementations > > > > Each accumulator will support two internal modes rather than introducing > > separate Ordered and Naive accumulator classes. > > > > > When the aggregation step is SINGLE and the planner can prove that samples > > within each group are ordered by time_col, the accumulator will use a > > streaming implementation with fixed-size state: > > > > Time complexity: O(n) > > > > Additional space per group: O(1) > > > > Otherwise, the accumulator will buffer all valid samples and sort them > > before the final calculation: > > > > Time complexity: O(n log n) > > > > Additional space per group: O(n) > > > > The two modes will have identical function semantics, validation rules, > > and results. > > > > A new inputOrderedByTimeAscending property will be added to > > AggregationNode.Aggregation. The ordered implementation will be selected > > only when: > > > > step == SINGLE && inputOrderedByTimeAscending > > > > PARTIAL, INTERMEDIATE, and FINAL aggregation stages will always use the > > buffered implementation because they need to generate, merge, or consume > > intermediate states. > > > > The ordered mode will still perform runtime checks for duplicate > > timestamps and unexpected out-of-order input. > > > > Accumulator organization > > > > rate() and increase() will share common base classes because they use the > > same counter-reset correction and boundary-extrapolation logic. > > > > The main TableAccumulator classes will include: > > > > AbstractRateIncreaseAccumulator > > > > RateAccumulator > > > > IncreaseAccumulator > > > > IrateAccumulator > > > > DeltaAccumulator > > > > Equivalent GroupedAccumulator classes will maintain state independently > > for each groupId. > > > > In ordered mode, the accumulators retain only fixed state such as the > > first, previous, and last samples, sample count, corrected increase, and > > window boundaries. > > > > In unordered mode, samples will be stored in a new TimeValueBuffer. > > Grouped accumulators will use TimeValueBufferBigArray to maintain one > > buffer per groupId. > > > > Intermediate-state handling > > > > For distributed aggregation, PARTIAL, INTERMEDIATE, and FINAL stages must > > preserve the complete sample sequence because counter-reset detection, > > > duplicate-timestamp validation, and the final result depend on global time > > ordering. > > > > The intermediate state will contain: > > > > A state version > > > > window_start and window_end, except for irate() > > > > The number of valid samples > > > > All timestamp/value pairs > > > > PARTIAL stages collect and serialize their samples without calculating a > > final result. > > > > INTERMEDIATE stages deserialize and merge sample buffers without sorting > > them. > > > > The FINAL stage merges all states, sorts the complete sample set by > > timestamp, checks for duplicate timestamps, and then performs the final > > calculation. > > > > Deserialization will validate the sample count and serialized length to > > > reject corrupted intermediate states and prevent invalid memory allocation. > > > > Memory management > > > > The unordered implementation will integrate with the existing query > > memory-management mechanism, following a design similar to the exact > > percentile() accumulator. > > > > TimeValueBuffer and TimeValueBufferBigArray will report their retained > > memory. Accumulators will use MemoryReservationManager to reserve or > > release the difference whenever buffers grow, merge, or shrink. > > > > > reset() will clear the calculation state, shrink oversized internal arrays > > to their initial capacity, and release the corresponding query memory. > > > > Shared extrapolation logic > > > > A shared ExtrapolationUtil will implement the boundary-extrapolation > > algorithm used by rate(), increase(), and delta(). It will be responsible > > for: > > > > Converting timestamp differences to seconds according to the current > > timestamp_precision > > > > Calculating the average sample interval > > > > Applying the 1.1-times extrapolation threshold > > > > Limiting extrapolation to half an average interval when a boundary is too > > far away > > > > Applying counter zero-point protection for rate() and increase() > > > > Calculating and validating the final extrapolation factor > > > > Counter-reset detection and sample traversal will remain in the > > accumulators rather than in ExtrapolationUtil. > > > > rate() will divide the extrapolated increase by the complete window > > duration. increase() will return the extrapolated increase directly. > > delta() will extrapolate the raw difference between the last and first > > values without counter zero-point protection. irate() will use only the > > final two samples and will not perform boundary extrapolation. > > > > All time conversions will respect the configured ms, us, or ns timestamp > > precision, while rate() and irate() will always return values per second. > > > > Please review this initial design, especially the ordered-input property, > > > the complete-sample intermediate state, and the memory-management approach. > > Any comments or alternative suggestions are welcome. > > > > Best regards, > > Xinqi Zhao > > > >
