Skip to content

Support rate family aggregation functions in table model #18334

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
CoollZzz wants to merge 1 commit into
base: master
Choose a base branch
from
Open

Support rate family aggregation functions in table model #18334

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
506 changes: 506 additions & 0 deletions ...st/src/test/java/org/apache/iotdb/relational/it/query/recent/IoTDBTableAggregationIT.java
View file
Open in desktop

Large diffs are not rendered by default.

57 changes: 57 additions & 0 deletions iotdb-core/calc-commons/src/main/i18n/en/org/apache/iotdb/calc/i18n/CalcMessages.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -468,5 +468,62 @@ private CalcMessages() {}
"Unsupported data type in PERCENTILE Aggregation: %s";
public static final String EXCEPTION_PERCENTILEACCUMULATOR_DOES_NOT_SUPPORT_STATISTICS_66308C79 =
"PercentileAccumulator does not support statistics";
public static final String
EXCEPTION_RATE_FAMILY_AGGREGATE_FUNCTION_REQUIRES_AT_LEAST_TWO_VALID_SAMPLES_1A89901C =
"Rate-family aggregate function requires at least two valid samples";
public static final String
EXCEPTION_THE_FIRST_SAMPLE_TIME_MUST_BE_LESS_THAN_THE_LAST_SAMPLE_TIME_FC4D3517 =
"The first sample time must be less than the last sample time";
public static final String
EXCEPTION_THE_WINDOW_START_MUST_BE_LESS_THAN_THE_WINDOW_END_B1A38C98 =
"The window start must be less than the window end";
public static final String
EXCEPTION_RATE_FAMILY_AGGREGATE_FUNCTION_REQUIRES_WINDOW_START_FIRST_TIME_LAST_TIME_WINDOW_END_60DE93C8 =
"Rate-family aggregate function requires 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-family aggregate function produced a non-finite intermediate result";
public static final String
EXCEPTION_RATE_FAMILY_AGGREGATE_FUNCTION_PRODUCED_A_NON_FINITE_TIME_INTERVAL_B26C4162 =
"Rate-family aggregate function produced a non-finite time interval";
public static final String
EXCEPTION_RATE_FAMILY_AGGREGATE_FUNCTION_PRODUCED_A_NON_FINITE_EXTRAPOLATION_RESULT_6482CF1D =
"Rate-family aggregate function produced a non-finite extrapolation result";
public static final String
EXCEPTION_AGGREGATE_FUNCTION_ARG_DOES_NOT_SUPPORT_DUPLICATE_TIME_COL_VALUES_IN_THE_SAME_AGGREGATION_GROUP_ARG_087A91BC =
"Aggregate function [%s] does not support duplicate time_col values in the same aggregation group: %d";
public static final String
EXCEPTION_AGGREGATE_FUNCTION_ARG_DOES_NOT_SUPPORT_NON_FINITE_VALUE_COL_ARG_AC2AAC62 =
"Aggregate function [%s] does not support non-finite 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 =
"The value_col argument of aggregate function [%s] must be a non-negative number, but got: %s";
public static final String
EXCEPTION_THE_ARGUMENT_ARG_OF_AGGREGATE_FUNCTION_ARG_MUST_NOT_BE_NULL_WHEN_VALUE_COL_IS_NOT_NULL_7F087E99 =
"The argument %d of aggregate function [%s] must not be NULL when value_col is not NULL";
public static final String
EXCEPTION_THE_WINDOW_START_ARGUMENT_OF_AGGREGATE_FUNCTION_ARG_MUST_BE_LESS_THAN_WINDOW_END_17D2A79A =
"The window_start argument of aggregate function [%s] must be less than window_end";
public static final String
EXCEPTION_THE_SAMPLE_TIME_OF_AGGREGATE_FUNCTION_ARG_MUST_SATISFY_WINDOW_START_TIME_COL_WINDOW_END_35014D15 =
"The sample time of aggregate function [%s] must satisfy 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 =
"Aggregate function [%s] requires consistent window boundaries in the same aggregation group: expected [%d, %d), but got [%d, %d)";
public static final String
EXCEPTION_AGGREGATE_FUNCTION_ARG_EXPECTED_TIME_COL_IN_STRICTLY_ASCENDING_ORDER_BUT_GOT_ARG_AFTER_ARG_9289E0F9 =
"Aggregate function [%s] expected time_col in strictly ascending order, but got %d after %d";
public static final String
EXCEPTION_ORDERED_AGGREGATE_FUNCTION_ARG_DOES_NOT_SUPPORT_INTERMEDIATE_STATE_6B4B2B1B =
"Ordered aggregate function [%s] does not support intermediate state";
public static final String
EXCEPTION_INVALID_INTERMEDIATE_STATE_FOR_AGGREGATE_FUNCTION_ARG_2999C30B =
"Invalid intermediate state for aggregate function [%s]";
public static final String
EXCEPTION_AGGREGATE_FUNCTION_ARG_DOES_NOT_SUPPORT_VALUE_TYPE_ARG_9DD7388D =
"Aggregate function [%s] does not support value type %s";
public static final String
EXCEPTION_AGGREGATE_FUNCTION_ARG_REQUIRES_ARG_ARGUMENTS_BUT_GOT_ARG_D6249DD2 =
"Aggregate function [%s] requires %d arguments, but got %d";

}
57 changes: 57 additions & 0 deletions iotdb-core/calc-commons/src/main/i18n/zh/org/apache/iotdb/calc/i18n/CalcMessages.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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";
public static final String
EXCEPTION_ORDERED_AGGREGATE_FUNCTION_ARG_DOES_NOT_SUPPORT_INTERMEDIATE_STATE_6B4B2B1B =
"有序聚合函数 [%s] 不支持 Intermediate State";
public static final String
EXCEPTION_INVALID_INTERMEDIATE_STATE_FOR_AGGREGATE_FUNCTION_ARG_2999C30B =
"聚合函数 [%s] 的 Intermediate State 非法";
public static final String
EXCEPTION_AGGREGATE_FUNCTION_ARG_DOES_NOT_SUPPORT_VALUE_TYPE_ARG_9DD7388D =
"聚合函数 [%s] 不支持值类型 %s";
public static final String
EXCEPTION_AGGREGATE_FUNCTION_ARG_REQUIRES_ARG_ARGUMENTS_BUT_GOT_ARG_D6249DD2 =
"聚合函数 [%s] 需要 %d 个参数，但实际传入 %d 个";
}
141 changes: 140 additions & 1 deletion ...pache/iotdb/calc/execution/operator/source/relational/aggregation/AccumulatorFactory.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -56,11 +56,28 @@
import org.apache.iotdb.calc.execution.operator.source.relational.aggregation.grouped.GroupedVarianceAccumulator;
import org.apache.iotdb.calc.execution.operator.source.relational.aggregation.grouped.IntGroupedApproxMostFrequentAccumulator;
import org.apache.iotdb.calc.execution.operator.source.relational.aggregation.grouped.LongGroupedApproxMostFrequentAccumulator;
import org.apache.iotdb.calc.execution.operator.source.relational.aggregation.grouped.rate.GroupedNaiveDeltaAccumulator;
import org.apache.iotdb.calc.execution.operator.source.relational.aggregation.grouped.rate.GroupedNaiveIncreaseAccumulator;
import org.apache.iotdb.calc.execution.operator.source.relational.aggregation.grouped.rate.GroupedNaiveIrateAccumulator;
import org.apache.iotdb.calc.execution.operator.source.relational.aggregation.grouped.rate.GroupedNaiveRateAccumulator;
import org.apache.iotdb.calc.execution.operator.source.relational.aggregation.grouped.rate.GroupedOrderedDeltaAccumulator;
import org.apache.iotdb.calc.execution.operator.source.relational.aggregation.grouped.rate.GroupedOrderedIncreaseAccumulator;
import org.apache.iotdb.calc.execution.operator.source.relational.aggregation.grouped.rate.GroupedOrderedIrateAccumulator;
import org.apache.iotdb.calc.execution.operator.source.relational.aggregation.grouped.rate.GroupedOrderedRateAccumulator;
import org.apache.iotdb.calc.execution.operator.source.relational.aggregation.rate.NaiveDeltaAccumulator;
import org.apache.iotdb.calc.execution.operator.source.relational.aggregation.rate.NaiveIncreaseAccumulator;
import org.apache.iotdb.calc.execution.operator.source.relational.aggregation.rate.NaiveIrateAccumulator;
import org.apache.iotdb.calc.execution.operator.source.relational.aggregation.rate.NaiveRateAccumulator;
import org.apache.iotdb.calc.execution.operator.source.relational.aggregation.rate.OrderedDeltaAccumulator;
import org.apache.iotdb.calc.execution.operator.source.relational.aggregation.rate.OrderedIncreaseAccumulator;
import org.apache.iotdb.calc.execution.operator.source.relational.aggregation.rate.OrderedIrateAccumulator;
import org.apache.iotdb.calc.execution.operator.source.relational.aggregation.rate.OrderedRateAccumulator;
import org.apache.iotdb.calc.i18n.CalcMessages;
import org.apache.iotdb.calc.plan.planner.memory.MemoryReservationManager;
import org.apache.iotdb.common.rpc.thrift.TAggregationType;
import org.apache.iotdb.commons.queryengine.execution.operator.source.relational.aggregation.grouped.UpdateMemory;
import org.apache.iotdb.commons.queryengine.execution.operator.source.relational.aggregation.grouped.hash.MarkDistinctHash;
import org.apache.iotdb.commons.queryengine.plan.relational.planner.node.AggregationNode;
import org.apache.iotdb.commons.queryengine.plan.relational.sql.ast.Expression;
import org.apache.iotdb.commons.queryengine.plan.relational.type.InternalTypeManager;
import org.apache.iotdb.commons.queryengine.plan.udf.TableUDFUtils;
Expand Down Expand Up @@ -142,6 +159,38 @@ public static TableAccumulator createAccumulator(
boolean distinct,
MemoryReservationManager memoryReservationManager,
@Nullable IoTDBLocal ioTDBLocal) {
return createAccumulator(
functionName,
aggregationType,
inputDataTypes,
inputExpressions,
inputAttributes,
ascending,
isAggTableScan,
timeColumnName,
measurementColumnNames,
distinct,
AggregationNode.Step.SINGLE,
false,
memoryReservationManager,
ioTDBLocal);
}

public static TableAccumulator createAccumulator(
String functionName,
TAggregationType aggregationType,
List<TSDataType> inputDataTypes,
List<Expression> inputExpressions,
Map<String, String> inputAttributes,
boolean ascending,
boolean isAggTableScan,
String timeColumnName,
Set<String> measurementColumnNames,
boolean distinct,
AggregationNode.Step step,
boolean inputOrderedByTimeAscending,
MemoryReservationManager memoryReservationManager,
@Nullable IoTDBLocal ioTDBLocal) {
TableAccumulator result;

// Input expression size of 1 indicates aggregation split has occurred and this is a final
Expand Down Expand Up @@ -201,7 +250,13 @@ public static TableAccumulator createAccumulator(
? new FirstAccumulator(inputDataTypes.get(0), isAggTableScan)
: new FirstDescAccumulator(inputDataTypes.get(0));
} else {
result = createBuiltinAccumulator(aggregationType, inputDataTypes, memoryReservationManager);
result =
createBuiltinAccumulator(
aggregationType,
inputDataTypes,
step,
inputOrderedByTimeAscending,
memoryReservationManager);
}

if (distinct) {
Expand Down Expand Up @@ -247,6 +302,32 @@ public static GroupedAccumulator createGroupedAccumulator(
boolean distinct,
MemoryReservationManager memoryReservationManager,
@Nullable IoTDBLocal ioTDBLocal) {
return createGroupedAccumulator(
functionName,
aggregationType,
inputDataTypes,
inputExpressions,
inputAttributes,
ascending,
distinct,
AggregationNode.Step.SINGLE,
false,
memoryReservationManager,
ioTDBLocal);
}

public static GroupedAccumulator createGroupedAccumulator(
String functionName,
TAggregationType aggregationType,
List<TSDataType> inputDataTypes,
List<Expression> inputExpressions,
Map<String, String> inputAttributes,
boolean ascending,
boolean distinct,
AggregationNode.Step step,
boolean inputOrderedByTimeAscending,
MemoryReservationManager memoryReservationManager,
@Nullable IoTDBLocal ioTDBLocal) {
GroupedAccumulator result;

if (aggregationType == TAggregationType.UDAF) {
Expand All @@ -261,6 +342,8 @@ public static GroupedAccumulator createGroupedAccumulator(
inputExpressions,
inputAttributes,
ascending,
step,
inputOrderedByTimeAscending,
memoryReservationManager);
}

Expand Down Expand Up @@ -317,7 +400,11 @@ private static GroupedAccumulator createBuiltinGroupedAccumulator(
List<Expression> inputExpressions,
Map<String, String> inputAttributes,
boolean ascending,
AggregationNode.Step step,
boolean inputOrderedByTimeAscending,
MemoryReservationManager memoryReservationManager) {
boolean useOrderedImplementation =
step == AggregationNode.Step.SINGLE && inputOrderedByTimeAscending;
switch (aggregationType) {
case COUNT:
return new GroupedCountAccumulator();
Expand Down Expand Up @@ -403,6 +490,22 @@ private static GroupedAccumulator createBuiltinGroupedAccumulator(
inputDataTypes.get(0), CentralMomentAccumulator.MomentType.KURTOSIS);
case PERCENTILE:
return new GroupedPercentileAccumulator(inputDataTypes.get(0), memoryReservationManager);
case RATE:
return useOrderedImplementation
? new GroupedOrderedRateAccumulator(inputDataTypes.get(0), memoryReservationManager)
: new GroupedNaiveRateAccumulator(inputDataTypes.get(0), memoryReservationManager);
case INCREASE:
return useOrderedImplementation
? new GroupedOrderedIncreaseAccumulator(inputDataTypes.get(0), memoryReservationManager)
: new GroupedNaiveIncreaseAccumulator(inputDataTypes.get(0), memoryReservationManager);
case IRATE:
return useOrderedImplementation
? new GroupedOrderedIrateAccumulator(inputDataTypes.get(0), memoryReservationManager)
: new GroupedNaiveIrateAccumulator(inputDataTypes.get(0), memoryReservationManager);
case DELTA:
return useOrderedImplementation
? new GroupedOrderedDeltaAccumulator(inputDataTypes.get(0), memoryReservationManager)
: new GroupedNaiveDeltaAccumulator(inputDataTypes.get(0), memoryReservationManager);
default:
throw new IllegalArgumentException(
CalcMessages.INVALID_AGGREGATION_FUNCTION + aggregationType);
Expand All @@ -413,6 +516,22 @@ public static TableAccumulator createBuiltinAccumulator(
TAggregationType aggregationType,
List<TSDataType> inputDataTypes,
MemoryReservationManager memoryReservationManager) {
return createBuiltinAccumulator(
aggregationType,
inputDataTypes,
AggregationNode.Step.SINGLE,
false,
memoryReservationManager);
}

public static TableAccumulator createBuiltinAccumulator(
TAggregationType aggregationType,
List<TSDataType> inputDataTypes,
AggregationNode.Step step,
boolean inputOrderedByTimeAscending,
MemoryReservationManager memoryReservationManager) {
boolean useOrderedImplementation =
step == AggregationNode.Step.SINGLE && inputOrderedByTimeAscending;
switch (aggregationType) {
case COUNT:
return new CountAccumulator();
Expand Down Expand Up @@ -499,6 +618,22 @@ public static TableAccumulator createBuiltinAccumulator(
inputDataTypes.get(0), CentralMomentAccumulator.MomentType.KURTOSIS);
case PERCENTILE:
return new PercentileAccumulator(inputDataTypes.get(0), memoryReservationManager);
case RATE:
return useOrderedImplementation
? new OrderedRateAccumulator(inputDataTypes.get(0))
: new NaiveRateAccumulator(inputDataTypes.get(0), memoryReservationManager);
case INCREASE:
return useOrderedImplementation
? new OrderedIncreaseAccumulator(inputDataTypes.get(0))
: new NaiveIncreaseAccumulator(inputDataTypes.get(0), memoryReservationManager);
case IRATE:
return useOrderedImplementation
? new OrderedIrateAccumulator(inputDataTypes.get(0))
: new NaiveIrateAccumulator(inputDataTypes.get(0), memoryReservationManager);
case DELTA:
return useOrderedImplementation
? new OrderedDeltaAccumulator(inputDataTypes.get(0))
: new NaiveDeltaAccumulator(inputDataTypes.get(0), memoryReservationManager);
default:
throw new IllegalArgumentException(
CalcMessages.INVALID_AGGREGATION_FUNCTION + aggregationType);
Expand Down Expand Up @@ -573,6 +708,10 @@ public static boolean isMultiInputAggregation(TAggregationType aggregationType)
case COVAR_SAMP:
case REGR_SLOPE:
case REGR_INTERCEPT:
case RATE:
case INCREASE:
case IRATE:
case DELTA:
return true;
default:
return false;
Expand Down
Loading
Loading