Skip to content

Commit d18b6dd

Browse files
FearfulTomcat27FearfulTomcat27
committed
Add PERCENTILE aggregation function and related validation
1 parent c4fc2e9 commit d18b6dd

11 files changed

Lines changed: 808 additions & 1 deletion

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
@@ -4345,6 +4345,42 @@ public void approxPercentileTest() {
43454345
DATABASE_NAME);
43464346
}
43474347

4348+
@Test
4349+
public void percentileTest() {
4350+
tableResultSetEqualTest(
4351+
"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",
4352+
buildHeaders(6),
4353+
new String[] {"2024-09-24T06:15:40.000Z,40,43000,37.5,43.0,2024-09-24T06:15:40.000Z,"},
4354+
DATABASE_NAME);
4355+
4356+
tableResultSetEqualTest(
4357+
"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",
4358+
new String[] {"time", "province", "_col2", "_col3", "_col4"},
4359+
new String[] {
4360+
"2024-09-24T06:15:30.000Z,beijing,2024-09-24T06:15:30.000Z,30,null,",
4361+
"2024-09-24T06:15:31.000Z,beijing,2024-09-24T06:15:31.000Z,null,31000,",
4362+
"2024-09-24T06:15:35.000Z,beijing,2024-09-24T06:15:35.000Z,null,35000,",
4363+
"2024-09-24T06:15:36.000Z,beijing,2024-09-24T06:15:36.000Z,36,null,",
4364+
"2024-09-24T06:15:40.000Z,beijing,2024-09-24T06:15:40.000Z,40,40000,",
4365+
"2024-09-24T06:15:41.000Z,beijing,2024-09-24T06:15:41.000Z,41,null,",
4366+
"2024-09-24T06:15:46.000Z,beijing,2024-09-24T06:15:46.000Z,null,46000,",
4367+
"2024-09-24T06:15:50.000Z,beijing,2024-09-24T06:15:50.000Z,null,50000,",
4368+
"2024-09-24T06:15:51.000Z,beijing,2024-09-24T06:15:51.000Z,null,null,",
4369+
"2024-09-24T06:15:55.000Z,beijing,2024-09-24T06:15:55.000Z,55,null,",
4370+
"2024-09-24T06:15:30.000Z,shanghai,2024-09-24T06:15:30.000Z,30,null,",
4371+
"2024-09-24T06:15:31.000Z,shanghai,2024-09-24T06:15:31.000Z,null,31000,",
4372+
"2024-09-24T06:15:35.000Z,shanghai,2024-09-24T06:15:35.000Z,null,35000,",
4373+
"2024-09-24T06:15:36.000Z,shanghai,2024-09-24T06:15:36.000Z,36,null,",
4374+
"2024-09-24T06:15:40.000Z,shanghai,2024-09-24T06:15:40.000Z,40,40000,",
4375+
"2024-09-24T06:15:41.000Z,shanghai,2024-09-24T06:15:41.000Z,41,null,",
4376+
"2024-09-24T06:15:46.000Z,shanghai,2024-09-24T06:15:46.000Z,null,46000,",
4377+
"2024-09-24T06:15:50.000Z,shanghai,2024-09-24T06:15:50.000Z,null,50000,",
4378+
"2024-09-24T06:15:51.000Z,shanghai,2024-09-24T06:15:51.000Z,null,null,",
4379+
"2024-09-24T06:15:55.000Z,shanghai,2024-09-24T06:15:55.000Z,55,null,",
4380+
},
4381+
DATABASE_NAME);
4382+
}
4383+
43484384
@Test
43494385
public void exceptionTest() {
43504386
tableAssertTestFail(
@@ -4423,6 +4459,22 @@ public void exceptionTest() {
44234459
"select approx_percentile(s5,0.5) from table1",
44244460
"701: Aggregation functions [approx_percentile] should have value column as numeric type [INT32, INT64, FLOAT, DOUBLE, TIMESTAMP]",
44254461
DATABASE_NAME);
4462+
tableAssertTestFail(
4463+
"select percentile() from table1",
4464+
"701: Aggregation functions [percentile] should only have two arguments",
4465+
DATABASE_NAME);
4466+
tableAssertTestFail(
4467+
"select percentile(s1,1.1) from table1",
4468+
"701: percentage should be in [0,1], got 1.1",
4469+
DATABASE_NAME);
4470+
tableAssertTestFail(
4471+
"select percentile(s1,'test') from table1",
4472+
"701: The second argument of 'percentile' function percentage must be a double literal",
4473+
DATABASE_NAME);
4474+
tableAssertTestFail(
4475+
"select percentile(s5,0.5) from table1",
4476+
"701: Aggregation functions [percentile] should have value column as numeric type [INT32, INT64, FLOAT, DOUBLE, TIMESTAMP]",
4477+
DATABASE_NAME);
44264478
}
44274479

44284480
// ==================================================================

‎iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/Percentile.java‎

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

‎iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/aggregation/AccumulatorFactory.java‎

Lines changed: 5 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -45,6 +45,7 @@
4545
import org.apache.iotdb.db.queryengine.execution.operator.source.relational.aggregation.grouped.GroupedMinAccumulator;
4646
import org.apache.iotdb.db.queryengine.execution.operator.source.relational.aggregation.grouped.GroupedMinByAccumulator;
4747
import org.apache.iotdb.db.queryengine.execution.operator.source.relational.aggregation.grouped.GroupedModeAccumulator;
48+
import org.apache.iotdb.db.queryengine.execution.operator.source.relational.aggregation.grouped.GroupedPercentileAccumulator;
4849
import org.apache.iotdb.db.queryengine.execution.operator.source.relational.aggregation.grouped.GroupedSumAccumulator;
4950
import org.apache.iotdb.db.queryengine.execution.operator.source.relational.aggregation.grouped.GroupedUserDefinedAggregateAccumulator;
5051
import org.apache.iotdb.db.queryengine.execution.operator.source.relational.aggregation.grouped.GroupedVarianceAccumulator;
@@ -283,6 +284,8 @@ private static GroupedAccumulator createBuiltinGroupedAccumulator(
283284
} else {
284285
return new GroupedApproxPercentileWithWeightAccumulator(inputDataTypes.get(0));
285286
}
287+
case PERCENTILE:
288+
return new GroupedPercentileAccumulator(inputDataTypes.get(0));
286289
default:
287290
throw new IllegalArgumentException("Invalid Aggregation function: " + aggregationType);
288291
}
@@ -367,6 +370,8 @@ public static TableAccumulator createBuiltinAccumulator(
367370
} else {
368371
return new ApproxPercentileWithWeightAccumulator(inputDataTypes.get(0));
369372
}
373+
case PERCENTILE:
374+
return new PercentileAccumulator(inputDataTypes.get(0));
370375
default:
371376
throw new IllegalArgumentException("Invalid Aggregation function: " + aggregationType);
372377
}

0 commit comments

Comments
 (0)