Skip to content

Commit aec7dfb

Browse files
FearfulTomcat27FearfulTomcat27
authored
Add PERCENTILE aggregation function and related validation (#16545)
1 parent 72e72dd commit aec7dfb

16 files changed

Lines changed: 985 additions & 32 deletions

File tree

‎integration-test/src/test/java/org/apache/iotdb/relational/it/query/recent/IoTDBTableAggregationIT.java‎

Lines changed: 52 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -4388,6 +4388,42 @@ public void approxPercentileTest() {
43884388
DATABASE_NAME);
43894389
}
43904390

4391+
@Test
4392+
public void percentileTest() {
4393+
tableResultSetEqualTest(
4394+
"select percentile(time, 0.5),percentile(s1,0.5),percentile(s2,0.5),percentile(s3,0.5),percentile(s4,0.5),percentile(s9,0.5) from table1",
4395+
buildHeaders(6),
4396+
new String[] {"2024-09-24T06:15:40.000Z,40,43000,37.5,43.0,2024-09-24T06:15:40.000Z,"},
4397+
DATABASE_NAME);
4398+
4399+
tableResultSetEqualTest(
4400+
"select time,province,percentile(time, 0.5),percentile(s1,0.5),percentile(s2,0.5) from table1 group by 1,2 order by 2,1",
4401+
new String[] {"time", "province", "_col2", "_col3", "_col4"},
4402+
new String[] {
4403+
"2024-09-24T06:15:30.000Z,beijing,2024-09-24T06:15:30.000Z,30,null,",
4404+
"2024-09-24T06:15:31.000Z,beijing,2024-09-24T06:15:31.000Z,null,31000,",
4405+
"2024-09-24T06:15:35.000Z,beijing,2024-09-24T06:15:35.000Z,null,35000,",
4406+
"2024-09-24T06:15:36.000Z,beijing,2024-09-24T06:15:36.000Z,36,null,",
4407+
"2024-09-24T06:15:40.000Z,beijing,2024-09-24T06:15:40.000Z,40,40000,",
4408+
"2024-09-24T06:15:41.000Z,beijing,2024-09-24T06:15:41.000Z,41,null,",
4409+
"2024-09-24T06:15:46.000Z,beijing,2024-09-24T06:15:46.000Z,null,46000,",
4410+
"2024-09-24T06:15:50.000Z,beijing,2024-09-24T06:15:50.000Z,null,50000,",
4411+
"2024-09-24T06:15:51.000Z,beijing,2024-09-24T06:15:51.000Z,null,null,",
4412+
"2024-09-24T06:15:55.000Z,beijing,2024-09-24T06:15:55.000Z,55,null,",
4413+
"2024-09-24T06:15:30.000Z,shanghai,2024-09-24T06:15:30.000Z,30,null,",
4414+
"2024-09-24T06:15:31.000Z,shanghai,2024-09-24T06:15:31.000Z,null,31000,",
4415+
"2024-09-24T06:15:35.000Z,shanghai,2024-09-24T06:15:35.000Z,null,35000,",
4416+
"2024-09-24T06:15:36.000Z,shanghai,2024-09-24T06:15:36.000Z,36,null,",
4417+
"2024-09-24T06:15:40.000Z,shanghai,2024-09-24T06:15:40.000Z,40,40000,",
4418+
"2024-09-24T06:15:41.000Z,shanghai,2024-09-24T06:15:41.000Z,41,null,",
4419+
"2024-09-24T06:15:46.000Z,shanghai,2024-09-24T06:15:46.000Z,null,46000,",
4420+
"2024-09-24T06:15:50.000Z,shanghai,2024-09-24T06:15:50.000Z,null,50000,",
4421+
"2024-09-24T06:15:51.000Z,shanghai,2024-09-24T06:15:51.000Z,null,null,",
4422+
"2024-09-24T06:15:55.000Z,shanghai,2024-09-24T06:15:55.000Z,55,null,",
4423+
},
4424+
DATABASE_NAME);
4425+
}
4426+
43914427
@Test
43924428
public void exceptionTest() {
43934429
tableAssertTestFail(
@@ -4478,6 +4514,22 @@ public void exceptionTest() {
44784514
"select 1 as g, approx_percentile(s1,s2,0.5) from table1 group by 1",
44794515
"701: Aggregation functions [approx_percentile] do not support weight as INT64 type",
44804516
DATABASE_NAME);
4517+
tableAssertTestFail(
4518+
"select percentile() from table1",
4519+
"701: Aggregation functions [percentile] should only have two arguments",
4520+
DATABASE_NAME);
4521+
tableAssertTestFail(
4522+
"select percentile(s1,1.1) from table1",
4523+
"701: percentage should be in [0,1], got 1.1",
4524+
DATABASE_NAME);
4525+
tableAssertTestFail(
4526+
"select percentile(s1,'test') from table1",
4527+
"701: The second argument of 'percentile' function percentage must be a double literal",
4528+
DATABASE_NAME);
4529+
tableAssertTestFail(
4530+
"select percentile(s5,0.5) from table1",
4531+
"701: Aggregation functions [percentile] should have value column as numeric type [INT32, INT64, FLOAT, DOUBLE, TIMESTAMP]",
4532+
DATABASE_NAME);
44814533
}
44824534

44834535
// ==================================================================

‎iotdb-core/calc-commons/src/main/java/org/apache/iotdb/calc/execution/operator/source/relational/Percentile.java‎

Lines changed: 173 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,173 @@
1+
/*
2+
* Licensed to the Apache Software Foundation (ASF) under one
3+
* or more contributor license agreements. See the NOTICE file
4+
* distributed with this work for additional information
5+
* regarding copyright ownership. The ASF licenses this file
6+
* to you under the Apache License, Version 2.0 (the
7+
* "License"); you may not use this file except in compliance
8+
* with the License. You may obtain a copy of the License at
9+
*
10+
* http://www.apache.org/licenses/LICENSE-2.0
11+
*
12+
* Unless required by applicable law or agreed to in writing,
13+
* software distributed under the License is distributed on an
14+
* "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
15+
* KIND, either express or implied. See the License for the
16+
* specific language governing permissions and limitations
17+
* under the License.
18+
*/
19+
20+
package org.apache.iotdb.calc.execution.operator.source.relational;
21+
22+
import org.apache.iotdb.commons.exception.SemanticException;
23+
24+
import org.apache.tsfile.utils.RamUsageEstimator;
25+
import org.apache.tsfile.utils.ReadWriteIOUtils;
26+
27+
import java.nio.ByteBuffer;
28+
import java.util.Arrays;
29+
30+
public class Percentile {
31+
private double[] values;
32+
private int size;
33+
private int capacity;
34+
private boolean sorted;
35+
36+
private static final int INITIAL_CAPACITY = 32;
37+
private static final double GROWTH_FACTOR = 1.5;
38+
39+
public Percentile() {
40+
this.capacity = INITIAL_CAPACITY;
41+
this.values = new double[capacity];
42+
this.size = 0;
43+
this.sorted = true;
44+
}
45+
46+
public void addValue(double value) {
47+
ensureCapacity();
48+
values[size++] = value;
49+
sorted = false;
50+
}
51+
52+
public void addValues(double... vals) {
53+
if (vals == null || vals.length == 0) {
54+
return;
55+
}
56+
57+
int newSize = size + vals.length;
58+
if (newSize > capacity) {
59+
grow(newSize);
60+
}
61+
62+
System.arraycopy(vals, 0, values, size, vals.length);
63+
size = newSize;
64+
sorted = false;
65+
}
66+
67+
public void merge(Percentile other) {
68+
if (other == null || other.size == 0) {
69+
return;
70+
}
71+
72+
int newSize = size + other.size;
73+
if (newSize > capacity) {
74+
grow(newSize);
75+
}
76+
77+
System.arraycopy(other.values, 0, values, size, other.size);
78+
size = newSize;
79+
sorted = false;
80+
}
81+
82+
public double getPercentile(double percentile) {
83+
if (size == 0) {
84+
return Double.NaN;
85+
}
86+
if (percentile < 0.0 || percentile > 1.0) {
87+
throw new SemanticException("percentage should be in [0,1], got " + percentile);
88+
}
89+
90+
ensureSorted();
91+
92+
if (size == 1) {
93+
return values[0];
94+
}
95+
96+
double realIndex = percentile * (size - 1);
97+
int index = (int) realIndex;
98+
double fraction = realIndex - index;
99+
100+
if (index >= size - 1) {
101+
return values[size - 1];
102+
}
103+
104+
return values[index] + fraction * (values[index + 1] - values[index]);
105+
}
106+
107+
public int getSize() {
108+
return size;
109+
}
110+
111+
public void clear() {
112+
// Shrink the backing array back to the initial capacity so the memory held by a large group is
113+
// actually released on reset, instead of staying reserved at the historical peak capacity.
114+
if (capacity > INITIAL_CAPACITY) {
115+
capacity = INITIAL_CAPACITY;
116+
values = new double[capacity];
117+
}
118+
size = 0;
119+
sorted = true;
120+
}
121+
122+
private void ensureCapacity() {
123+
if (size >= capacity) {
124+
grow(size + 1);
125+
}
126+
}
127+
128+
private void grow(int minCapacity) {
129+
int newCapacity = Math.max((int) (capacity * GROWTH_FACTOR), minCapacity);
130+
double[] newValues = new double[newCapacity];
131+
System.arraycopy(values, 0, newValues, 0, size);
132+
values = newValues;
133+
capacity = newCapacity;
134+
}
135+
136+
private void ensureSorted() {
137+
if (!sorted && size > 1) {
138+
Arrays.sort(values, 0, size);
139+
sorted = true;
140+
}
141+
}
142+
143+
public void serialize(ByteBuffer buffer) {
144+
ReadWriteIOUtils.write(size, buffer);
145+
for (int i = 0; i < size; i++) {
146+
ReadWriteIOUtils.write(values[i], buffer);
147+
}
148+
}
149+
150+
public static Percentile deserialize(ByteBuffer buffer) {
151+
int size = ReadWriteIOUtils.readInt(buffer);
152+
Percentile percentile = new Percentile();
153+
if (size > percentile.capacity) {
154+
percentile.capacity = size;
155+
percentile.values = new double[size];
156+
}
157+
percentile.size = size;
158+
for (int i = 0; i < size; i++) {
159+
percentile.values[i] = ReadWriteIOUtils.readDouble(buffer);
160+
}
161+
percentile.sorted = false;
162+
return percentile;
163+
}
164+
165+
public int getSerializedSize() {
166+
return Integer.BYTES + (int) ((long) size * Double.BYTES);
167+
}
168+
169+
public long getEstimatedSize() {
170+
return RamUsageEstimator.shallowSizeOfInstance(Percentile.class)
171+
+ (long) capacity * Double.BYTES;
172+
}
173+
}

‎iotdb-core/calc-commons/src/main/java/org/apache/iotdb/calc/execution/operator/source/relational/aggregation/AccumulatorFactory.java‎

Lines changed: 22 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -49,13 +49,15 @@
4949
import org.apache.iotdb.calc.execution.operator.source.relational.aggregation.grouped.GroupedMinAccumulator;
5050
import org.apache.iotdb.calc.execution.operator.source.relational.aggregation.grouped.GroupedMinByAccumulator;
5151
import org.apache.iotdb.calc.execution.operator.source.relational.aggregation.grouped.GroupedModeAccumulator;
52+
import org.apache.iotdb.calc.execution.operator.source.relational.aggregation.grouped.GroupedPercentileAccumulator;
5253
import org.apache.iotdb.calc.execution.operator.source.relational.aggregation.grouped.GroupedRegressionAccumulator;
5354
import org.apache.iotdb.calc.execution.operator.source.relational.aggregation.grouped.GroupedSumAccumulator;
5455
import org.apache.iotdb.calc.execution.operator.source.relational.aggregation.grouped.GroupedUserDefinedAggregateAccumulator;
5556
import org.apache.iotdb.calc.execution.operator.source.relational.aggregation.grouped.GroupedVarianceAccumulator;
5657
import org.apache.iotdb.calc.execution.operator.source.relational.aggregation.grouped.IntGroupedApproxMostFrequentAccumulator;
5758
import org.apache.iotdb.calc.execution.operator.source.relational.aggregation.grouped.LongGroupedApproxMostFrequentAccumulator;
5859
import org.apache.iotdb.calc.i18n.CalcMessages;
60+
import org.apache.iotdb.calc.plan.planner.memory.MemoryReservationManager;
5961
import org.apache.iotdb.common.rpc.thrift.TAggregationType;
6062
import org.apache.iotdb.commons.queryengine.execution.operator.source.relational.aggregation.grouped.UpdateMemory;
6163
import org.apache.iotdb.commons.queryengine.execution.operator.source.relational.aggregation.grouped.hash.MarkDistinctHash;
@@ -106,7 +108,8 @@ public static TableAccumulator createAccumulator(
106108
boolean isAggTableScan,
107109
String timeColumnName,
108110
Set<String> measurementColumnNames,
109-
boolean distinct) {
111+
boolean distinct,
112+
MemoryReservationManager memoryReservationManager) {
110113
TableAccumulator result;
111114

112115
// Input expression size of 1 indicates aggregation split has occurred and this is a final
@@ -166,7 +169,7 @@ public static TableAccumulator createAccumulator(
166169
? new FirstAccumulator(inputDataTypes.get(0), isAggTableScan)
167170
: new FirstDescAccumulator(inputDataTypes.get(0));
168171
} else {
169-
result = createBuiltinAccumulator(aggregationType, inputDataTypes);
172+
result = createBuiltinAccumulator(aggregationType, inputDataTypes, memoryReservationManager);
170173
}
171174

172175
if (distinct) {
@@ -188,7 +191,8 @@ public static GroupedAccumulator createGroupedAccumulator(
188191
List<Expression> inputExpressions,
189192
Map<String, String> inputAttributes,
190193
boolean ascending,
191-
boolean distinct) {
194+
boolean distinct,
195+
MemoryReservationManager memoryReservationManager) {
192196
GroupedAccumulator result;
193197

194198
if (aggregationType == TAggregationType.UDAF) {
@@ -197,7 +201,12 @@ public static GroupedAccumulator createGroupedAccumulator(
197201
} else {
198202
result =
199203
createBuiltinGroupedAccumulator(
200-
aggregationType, inputDataTypes, inputExpressions, inputAttributes, ascending);
204+
aggregationType,
205+
inputDataTypes,
206+
inputExpressions,
207+
inputAttributes,
208+
ascending,
209+
memoryReservationManager);
201210
}
202211

203212
if (distinct) {
@@ -242,7 +251,8 @@ private static GroupedAccumulator createBuiltinGroupedAccumulator(
242251
List<TSDataType> inputDataTypes,
243252
List<Expression> inputExpressions,
244253
Map<String, String> inputAttributes,
245-
boolean ascending) {
254+
boolean ascending,
255+
MemoryReservationManager memoryReservationManager) {
246256
switch (aggregationType) {
247257
case COUNT:
248258
return new GroupedCountAccumulator();
@@ -326,14 +336,18 @@ private static GroupedAccumulator createBuiltinGroupedAccumulator(
326336
case KURTOSIS:
327337
return new GroupedCentralMomentAccumulator(
328338
inputDataTypes.get(0), CentralMomentAccumulator.MomentType.KURTOSIS);
339+
case PERCENTILE:
340+
return new GroupedPercentileAccumulator(inputDataTypes.get(0), memoryReservationManager);
329341
default:
330342
throw new IllegalArgumentException(
331343
CalcMessages.INVALID_AGGREGATION_FUNCTION + aggregationType);
332344
}
333345
}
334346

335347
public static TableAccumulator createBuiltinAccumulator(
336-
TAggregationType aggregationType, List<TSDataType> inputDataTypes) {
348+
TAggregationType aggregationType,
349+
List<TSDataType> inputDataTypes,
350+
MemoryReservationManager memoryReservationManager) {
337351
switch (aggregationType) {
338352
case COUNT:
339353
return new CountAccumulator();
@@ -418,6 +432,8 @@ public static TableAccumulator createBuiltinAccumulator(
418432
case KURTOSIS:
419433
return new TableCentralMomentAccumulator(
420434
inputDataTypes.get(0), CentralMomentAccumulator.MomentType.KURTOSIS);
435+
case PERCENTILE:
436+
return new PercentileAccumulator(inputDataTypes.get(0), memoryReservationManager);
421437
default:
422438
throw new IllegalArgumentException(
423439
CalcMessages.INVALID_AGGREGATION_FUNCTION + aggregationType);

0 commit comments

Comments
 (0)