JackieTien97 commented on code in PR #18334: URL: https://github.com/apache/iotdb/pull/18334#discussion_r3688754137
########## iotdb-core/calc-commons/src/main/java/org/apache/iotdb/calc/execution/operator/source/relational/aggregation/rate/ExtrapolationUtil.java: ########## @@ -0,0 +1,121 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package org.apache.iotdb.calc.execution.operator.source.relational.aggregation.rate; + +import org.apache.iotdb.calc.i18n.CalcMessages; +import org.apache.iotdb.commons.exception.SemanticException; +import org.apache.iotdb.commons.queryengine.utils.TimestampPrecisionUtils; + +import java.math.BigInteger; +import java.util.concurrent.TimeUnit; + +public final class ExtrapolationUtil { + + private static final double EXTRAPOLATION_THRESHOLD_FACTOR = 1.1; + + private ExtrapolationUtil() {} + + public static double extrapolate( + int sampleCount, + long firstTime, + double firstValue, + long lastTime, + long windowStart, + long windowEnd, + double increase, + boolean applyCounterZeroProtection) { + validateBoundaries(sampleCount, firstTime, lastTime, windowStart, windowEnd); + + double sampledInterval = timestampDiffToSeconds(lastTime, firstTime); + double durationToStart = timestampDiffToSeconds(firstTime, windowStart); + double durationToEnd = timestampDiffToSeconds(windowEnd, lastTime); + + if (!Double.isFinite(increase)) { + throw new SemanticException( + CalcMessages + .EXCEPTION_RATE_FAMILY_AGGREGATE_FUNCTION_PRODUCED_A_NON_FINITE_INTERMEDIATE_RESULT_D46B30CD); + } + if (applyCounterZeroProtection && increase == 0.0) { + return 0.0; + } + + double averageInterval = sampledInterval / (sampleCount - 1); + double threshold = averageInterval * EXTRAPOLATION_THRESHOLD_FACTOR; + + if (durationToStart >= threshold) { Review Comment: [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. ########## integration-test/src/test/java/org/apache/iotdb/relational/it/query/recent/IoTDBTableAggregationIT.java: ########## @@ -5717,6 +5750,479 @@ public void orderByLimitTest() { DATABASE_NAME); } + // ================================================================== + // ================= Rate-Family Aggregation Tests ================== + // ================================================================== + + @Test + public void rateFunctionsNormalTest() { + tableResultSetEqualTest( + "SELECT scenario,date_bin(5ms,time) AS window_start," + + "rate(value_double,time,window_start,window_start + 5) AS rate_value," + + "increase(value_double,time,window_start,window_start + 5) AS increase_value," + + "irate(value_double,time) AS irate_value," + + "delta(value_double,time,window_start,window_start + 5) AS delta_value " + + "FROM (SELECT time,scenario,value_double FROM rate_test " + + "WHERE scenario='timestamp_normal') GROUP BY scenario,window_start", + new String[] { + "scenario", "window_start", "rate_value", "increase_value", "irate_value", "delta_value" + }, + new String[] {"timestamp_normal,1970-01-01T00:00:00.000Z,60000.0,300.0,60000.0,300.0,"}, + DATABASE_NAME); + + tableResultSetEqualTest( + "SELECT rate(value_double,sample_time,window_start,window_end) AS rate_value," + + "increase(value_double,sample_time,window_start,window_end) AS increase_value," + + "irate(value_double,sample_time) AS irate_value," + + "delta(value_double,sample_time,window_start,window_end) AS delta_value " + + "FROM rate_test WHERE scenario='reset'", + new String[] {"rate_value", "increase_value", "irate_value", "delta_value"}, + new String[] {"65.0,325.0,80.0,-400.0,"}, + DATABASE_NAME); + + tableResultSetEqualTest( + "SELECT scenario,rate(value_double,sample_time,window_start,window_end) AS rate_value," + + "increase(value_double,sample_time,window_start,window_end) AS increase_value," + + "irate(value_double,sample_time) AS irate_value," + + "delta(value_double,sample_time,window_start,window_end) AS delta_value " + + "FROM rate_test WHERE scenario IN ('timestamp_normal','reset') " + + "GROUP BY scenario ORDER BY scenario", + new String[] {"scenario", "rate_value", "increase_value", "irate_value", "delta_value"}, + new String[] {"reset,65.0,325.0,80.0,-400.0,", "timestamp_normal,60.0,300.0,60.0,300.0,"}, + DATABASE_NAME); + + // The rows span two flushed files and two physical time partitions. In cluster mode this also + // exercises partial/intermediate/final aggregation. Review Comment: [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. ########## iotdb-core/calc-commons/src/main/java/org/apache/iotdb/calc/execution/operator/source/relational/aggregation/rate/RateFunctionIntermediateStateCodec.java: ########## @@ -0,0 +1,143 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package org.apache.iotdb.calc.execution.operator.source.relational.aggregation.rate; + +import org.apache.iotdb.calc.i18n.CalcMessages; +import org.apache.iotdb.commons.exception.SemanticException; + +import org.apache.tsfile.block.column.ColumnBuilder; +import org.apache.tsfile.utils.Binary; + +import java.nio.ByteBuffer; + +public final class RateFunctionIntermediateStateCodec { + + private static final int STATE_VERSION = 1; + private static final int WINDOWED_HEADER_SIZE = 24; + private static final int IRATE_HEADER_SIZE = 8; + private static final int SAMPLE_SIZE = 16; + + private RateFunctionIntermediateStateCodec() {} + + public static void encode( + RateFunctionType functionType, + long windowStart, + long windowEnd, + TimeValueBuffer samples, + ColumnBuilder output) { + if (samples == null || samples.isEmpty()) { + output.appendNull(); + return; + } + + long headerSize = functionType.isWindowed() ? WINDOWED_HEADER_SIZE : IRATE_HEADER_SIZE; + long serializedSize = + Math.addExact(headerSize, Math.multiplyExact((long) samples.size(), SAMPLE_SIZE)); + ByteBuffer target = ByteBuffer.allocate(Math.toIntExact(serializedSize)); Review Comment: [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. ########## iotdb-core/calc-commons/src/main/i18n/zh/org/apache/iotdb/calc/i18n/CalcMessages.java: ########## @@ -445,4 +445,61 @@ private CalcMessages() {} "PERCENTILE 聚合中不支持的数据类型：%s"; public static final String EXCEPTION_PERCENTILEACCUMULATOR_DOES_NOT_SUPPORT_STATISTICS_66308C79 = "PercentileAccumulator 不支持统计信息"; + public static final String + EXCEPTION_RATE_FAMILY_AGGREGATE_FUNCTION_REQUIRES_AT_LEAST_TWO_VALID_SAMPLES_1A89901C = + "rate 系列聚合函数至少需要两个有效样本"; + public static final String + EXCEPTION_THE_FIRST_SAMPLE_TIME_MUST_BE_LESS_THAN_THE_LAST_SAMPLE_TIME_FC4D3517 = + "第一个样本时间必须小于最后一个样本时间"; + public static final String + EXCEPTION_THE_WINDOW_START_MUST_BE_LESS_THAN_THE_WINDOW_END_B1A38C98 = + "window_start 必须小于 window_end"; + public static final String + EXCEPTION_RATE_FAMILY_AGGREGATE_FUNCTION_REQUIRES_WINDOW_START_FIRST_TIME_LAST_TIME_WINDOW_END_60DE93C8 = + "rate 系列聚合函数要求 window_start <= first_time < last_time < window_end"; + public static final String + EXCEPTION_RATE_FAMILY_AGGREGATE_FUNCTION_PRODUCED_A_NON_FINITE_INTERMEDIATE_RESULT_D46B30CD = + "rate 系列聚合函数产生了非有限中间结果"; + public static final String + EXCEPTION_RATE_FAMILY_AGGREGATE_FUNCTION_PRODUCED_A_NON_FINITE_TIME_INTERVAL_B26C4162 = + "rate 系列聚合函数产生了非有限时间间隔"; + public static final String + EXCEPTION_RATE_FAMILY_AGGREGATE_FUNCTION_PRODUCED_A_NON_FINITE_EXTRAPOLATION_RESULT_6482CF1D = + "rate 系列聚合函数产生了非有限外推结果"; + public static final String + EXCEPTION_AGGREGATE_FUNCTION_ARG_DOES_NOT_SUPPORT_DUPLICATE_TIME_COL_VALUES_IN_THE_SAME_AGGREGATION_GROUP_ARG_087A91BC = + "聚合函数 [%s] 不支持同一聚合分组内出现重复的 time_col 值：%d"; + public static final String + EXCEPTION_AGGREGATE_FUNCTION_ARG_DOES_NOT_SUPPORT_NON_FINITE_VALUE_COL_ARG_AC2AAC62 = + "聚合函数 [%s] 不支持非有限 value_col：%s"; + public static final String + EXCEPTION_THE_VALUE_COL_ARGUMENT_OF_AGGREGATE_FUNCTION_ARG_MUST_BE_A_NON_NEGATIVE_NUMBER_BUT_GOT_ARG_4D5B7D74 = + "聚合函数 [%s] 的 value_col 参数必须是非负数，实际为：%s"; + public static final String + EXCEPTION_THE_ARGUMENT_ARG_OF_AGGREGATE_FUNCTION_ARG_MUST_NOT_BE_NULL_WHEN_VALUE_COL_IS_NOT_NULL_7F087E99 = + "第 %d 个参数在聚合函数 [%s] 的 value_col 不为 NULL 时不得为 NULL"; + public static final String + EXCEPTION_THE_WINDOW_START_ARGUMENT_OF_AGGREGATE_FUNCTION_ARG_MUST_BE_LESS_THAN_WINDOW_END_17D2A79A = + "聚合函数 [%s] 的 window_start 参数必须小于 window_end"; + public static final String + EXCEPTION_THE_SAMPLE_TIME_OF_AGGREGATE_FUNCTION_ARG_MUST_SATISFY_WINDOW_START_TIME_COL_WINDOW_END_35014D15 = + "聚合函数 [%s] 的样本时间必须满足 window_start <= time_col < window_end"; + public static final String + EXCEPTION_AGGREGATE_FUNCTION_ARG_REQUIRES_CONSISTENT_WINDOW_BOUNDARIES_IN_THE_SAME_AGGREGATION_GROUP_EXPECTED_ARG_ARG_BUT_GOT_ARG_ARG_38631886 = + "聚合函数 [%s] 要求同一聚合分组内的窗口边界一致：期望 [%d, %d)，实际为 [%d, %d)"; + public static final String + EXCEPTION_AGGREGATE_FUNCTION_ARG_EXPECTED_TIME_COL_IN_STRICTLY_ASCENDING_ORDER_BUT_GOT_ARG_AFTER_ARG_9289E0F9 = + "聚合函数 [%s] 要求 time_col 严格升序，但在 %d 之后得到了 %d"; Review Comment: [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. ########## iotdb-core/calc-commons/src/main/java/org/apache/iotdb/calc/execution/operator/source/relational/aggregation/grouped/rate/GroupedNaiveRateAccumulator.java: ########## @@ -0,0 +1,163 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package org.apache.iotdb.calc.execution.operator.source.relational.aggregation.grouped.rate; + +import org.apache.iotdb.calc.execution.operator.source.relational.aggregation.AggregationMask; +import org.apache.iotdb.calc.execution.operator.source.relational.aggregation.rate.RateFunctionIntermediateStateCodec; +import org.apache.iotdb.calc.execution.operator.source.relational.aggregation.rate.RateFunctionType; +import org.apache.iotdb.calc.execution.operator.source.relational.aggregation.rate.RateFunctionValidation; +import org.apache.iotdb.calc.execution.operator.source.relational.aggregation.rate.TimeValueBuffer; +import org.apache.iotdb.calc.plan.planner.memory.MemoryReservationManager; + +import org.apache.tsfile.block.column.Column; +import org.apache.tsfile.block.column.ColumnBuilder; +import org.apache.tsfile.enums.TSDataType; +import org.apache.tsfile.utils.RamUsageEstimator; + +public final class GroupedNaiveRateAccumulator extends AbstractGroupedRateAccumulator { + + private static final long INSTANCE_SIZE = + RamUsageEstimator.shallowSizeOfInstance(GroupedNaiveRateAccumulator.class); + + private final TimeValueBufferBigArray samples = new TimeValueBufferBigArray(); + private final MemoryReservationManager memoryReservationManager; + private long previousSize; + + public GroupedNaiveRateAccumulator( + TSDataType valueDataType, MemoryReservationManager memoryReservationManager) { + super(valueDataType); + this.memoryReservationManager = memoryReservationManager; + updateMemoryReservation(); + } + + @Override + public long getEstimatedSize() { + return INSTANCE_SIZE + windowsSizeOf() + samples.sizeOf(); + } + + @Override + public void setGroupCount(long groupCount) { + ensureWindowCapacity(groupCount); + samples.ensureCapacity(groupCount); + updateMemoryReservation(); + } + + @Override + public void addInput(int[] groupIds, Column[] arguments, AggregationMask mask) { + RateFunctionValidation.validateArgumentCount(arguments, RateFunctionType.RATE); + int selectedCount = mask.getSelectedPositionCount(); + int[] selectedPositions = mask.isSelectAll() ? null : mask.getSelectedPositions(); + for (int index = 0; index < selectedCount; index++) { + int position = mask.isSelectAll() ? index : selectedPositions[index]; + if (arguments[0].isNull(position)) { + continue; + } + int groupId = groupIds[position]; + double value = + RateFunctionValidation.readValue( + arguments[0], position, valueDataType, RateFunctionType.RATE); + long time = + RateFunctionValidation.readRequiredTime(arguments[1], position, RateFunctionType.RATE, 2); + long currentWindowStart = + RateFunctionValidation.readRequiredTime(arguments[2], position, RateFunctionType.RATE, 3); + long currentWindowEnd = + RateFunctionValidation.readRequiredTime(arguments[3], position, RateFunctionType.RATE, 4); + RateFunctionValidation.validateWindow( + RateFunctionType.RATE, time, currentWindowStart, currentWindowEnd); + initializeOrValidateWindow(groupId, currentWindowStart, currentWindowEnd); + samples.add(groupId, time, value); + } + updateMemoryReservation(); + } + + @Override + public void addIntermediate(int[] groupIds, Column argument) { + for (int position = 0; position < groupIds.length; position++) { + if (argument.isNull(position)) { + continue; + } + int groupId = groupIds[position]; + RateFunctionIntermediateStateCodec.DecodedState decoded = + RateFunctionIntermediateStateCodec.decode( + RateFunctionType.RATE, argument.getBinary(position)); + initializeOrValidateWindow(groupId, decoded.getWindowStart(), decoded.getWindowEnd()); + samples.merge(groupId, decoded.getSamples()); + } + updateMemoryReservation(); + } + + @Override + public void evaluateIntermediate(int groupId, ColumnBuilder output) { + RateFunctionIntermediateStateCodec.encode( + RateFunctionType.RATE, + windowStart(groupId), + windowEnd(groupId), + samples.get(groupId), + output); + } + + @Override + public void evaluateFinal(int groupId, ColumnBuilder output) { + TimeValueBuffer buffer = samples.get(groupId); + if (buffer == null || buffer.size() < 2) { + output.appendNull(); + return; + } + buffer.sortAndValidate(RateFunctionType.RATE.getFunctionName()); + int lastIndex = buffer.size() - 1; + output.writeDouble( + calculateRate( + buffer.size(), + buffer.getTime(0), + buffer.getValue(0), + buffer.getTime(lastIndex), + calculateCorrectedIncrease(buffer), + windowStart(groupId), + windowEnd(groupId))); + } + + @Override + public void reset() { + resetWindows(); + samples.reset(); + updateMemoryReservation(); + } + + private double calculateCorrectedIncrease(TimeValueBuffer buffer) { + double result = 0.0; + for (int index = 1; index < buffer.size(); index++) { + double previous = buffer.getValue(index - 1); + double current = buffer.getValue(index); + result = validateFinite(result + (current >= previous ? current - previous : current)); + } + return result; + } + + private void updateMemoryReservation() { + long currentSize = getEstimatedSize(); + long delta = currentSize - previousSize; + if (delta > 0) { + memoryReservationManager.reserveMemoryCumulatively(delta); Review Comment: [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. ########## integration-test/src/test/java/org/apache/iotdb/relational/it/query/recent/IoTDBTableAggregationIT.java: ########## @@ -5717,6 +5750,479 @@ public void orderByLimitTest() { DATABASE_NAME); } + // ================================================================== + // ================= Rate-Family Aggregation Tests ================== + // ================================================================== + + @Test + public void rateFunctionsNormalTest() { + tableResultSetEqualTest( + "SELECT scenario,date_bin(5ms,time) AS window_start," + + "rate(value_double,time,window_start,window_start + 5) AS rate_value," + + "increase(value_double,time,window_start,window_start + 5) AS increase_value," + + "irate(value_double,time) AS irate_value," + + "delta(value_double,time,window_start,window_start + 5) AS delta_value " + + "FROM (SELECT time,scenario,value_double FROM rate_test " + + "WHERE scenario='timestamp_normal') GROUP BY scenario,window_start", + new String[] { + "scenario", "window_start", "rate_value", "increase_value", "irate_value", "delta_value" + }, + new String[] {"timestamp_normal,1970-01-01T00:00:00.000Z,60000.0,300.0,60000.0,300.0,"}, + DATABASE_NAME); + + tableResultSetEqualTest( + "SELECT rate(value_double,sample_time,window_start,window_end) AS rate_value," + + "increase(value_double,sample_time,window_start,window_end) AS increase_value," + + "irate(value_double,sample_time) AS irate_value," + + "delta(value_double,sample_time,window_start,window_end) AS delta_value " + + "FROM rate_test WHERE scenario='reset'", + new String[] {"rate_value", "increase_value", "irate_value", "delta_value"}, + new String[] {"65.0,325.0,80.0,-400.0,"}, + DATABASE_NAME); + + tableResultSetEqualTest( + "SELECT scenario,rate(value_double,sample_time,window_start,window_end) AS rate_value," + + "increase(value_double,sample_time,window_start,window_end) AS increase_value," + + "irate(value_double,sample_time) AS irate_value," + + "delta(value_double,sample_time,window_start,window_end) AS delta_value " + + "FROM rate_test WHERE scenario IN ('timestamp_normal','reset') " + + "GROUP BY scenario ORDER BY scenario", + new String[] {"scenario", "rate_value", "increase_value", "irate_value", "delta_value"}, + new String[] {"reset,65.0,325.0,80.0,-400.0,", "timestamp_normal,60.0,300.0,60.0,300.0,"}, + DATABASE_NAME); + + // The rows span two flushed files and two physical time partitions. In cluster mode this also + // exercises partial/intermediate/final aggregation. + tableResultSetEqualTest( + "SELECT rate(value_double,sample_time,window_start,window_end) AS rate_value," + + "increase(value_double,sample_time,window_start,window_end) AS increase_value," + + "irate(value_double,sample_time) AS irate_value," + + "delta(gauge,sample_time,window_start,window_end) AS delta_value " + + "FROM rate_test WHERE scenario='distributed'", + new String[] {"rate_value", "increase_value", "irate_value", "delta_value"}, + new String[] {"60.0,240.0,80.0,-40.0,"}, + DATABASE_NAME); + + tableResultSetEqualTest( + "SELECT rate(value_double,sample_time,window_start,window_end) AS rate_value," + + "increase(value_double,sample_time,window_start,window_end) AS increase_value," + + "irate(value_double,sample_time) AS irate_value," + + "delta(value_double,sample_time,window_start,window_end) AS delta_value " + + "FROM rate_test WHERE scenario='unordered'", + new String[] {"rate_value", "increase_value", "irate_value", "delta_value"}, + new String[] {"10.0,50.0,10.0,50.0,"}, + DATABASE_NAME); + + tableResultSetEqualTest( + "SELECT window_start,window_end," + + "rate(value_double,time,window_start,window_end) AS rate_value," + + "increase(value_double,time,window_start,window_end) AS increase_value," + + "irate(value_double,time) AS irate_value," + + "delta(value_double,time,window_start,window_end) AS delta_value " + + "FROM TUMBLE(DATA => (SELECT time,value_double FROM rate_test " + + "WHERE scenario='timestamp_normal'),TIMECOL => 'time',SIZE => 5ms) " + + "GROUP BY window_start,window_end", + new String[] { + "window_start", "window_end", "rate_value", "increase_value", "irate_value", "delta_value" + }, + new String[] { + "1970-01-01T00:00:00.000Z,1970-01-01T00:00:00.005Z,60000.0,300.0,60000.0,300.0," + }, + DATABASE_NAME); + + tableResultSetEqualTest( + "SELECT window_start,window_end," + + "rate(value_double,time,window_start,window_end) AS rate_value " + + "FROM HOP(DATA => (SELECT time,value_double FROM rate_test " + + "WHERE scenario='timestamp_normal'),TIMECOL => 'time',SLIDE => 5ms,SIZE => 5ms) " Review Comment: [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. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
