Skip to content

Commit f63a796

Browse files
FearfulTomcat27FearfulTomcat27
committed
feat: Add PERCENTILE aggregation function with validation and error handling
1 parent 112bcd8 commit f63a796

11 files changed

Lines changed: 801 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
@@ -4366,6 +4366,42 @@ public void approxPercentileTest() {
43664366
DATABASE_NAME);
43674367
}
43684368

4369+
@Test
4370+
public void percentileTest() {
4371+
tableResultSetEqualTest(
4372+
"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",
4373+
buildHeaders(6),
4374+
new String[] {"2024-09-24T06:15:40.000Z,40,43000,37.5,43.0,2024-09-24T06:15:40.000Z,"},
4375+
DATABASE_NAME);
4376+
4377+
tableResultSetEqualTest(
4378+
"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",
4379+
new String[] {"time", "province", "_col2", "_col3", "_col4"},
4380+
new String[] {
4381+
"2024-09-24T06:15:30.000Z,beijing,2024-09-24T06:15:30.000Z,30,null,",
4382+
"2024-09-24T06:15:31.000Z,beijing,2024-09-24T06:15:31.000Z,null,31000,",
4383+
"2024-09-24T06:15:35.000Z,beijing,2024-09-24T06:15:35.000Z,null,35000,",
4384+
"2024-09-24T06:15:36.000Z,beijing,2024-09-24T06:15:36.000Z,36,null,",
4385+
"2024-09-24T06:15:40.000Z,beijing,2024-09-24T06:15:40.000Z,40,40000,",
4386+
"2024-09-24T06:15:41.000Z,beijing,2024-09-24T06:15:41.000Z,41,null,",
4387+
"2024-09-24T06:15:46.000Z,beijing,2024-09-24T06:15:46.000Z,null,46000,",
4388+
"2024-09-24T06:15:50.000Z,beijing,2024-09-24T06:15:50.000Z,null,50000,",
4389+
"2024-09-24T06:15:51.000Z,beijing,2024-09-24T06:15:51.000Z,null,null,",
4390+
"2024-09-24T06:15:55.000Z,beijing,2024-09-24T06:15:55.000Z,55,null,",
4391+
"2024-09-24T06:15:30.000Z,shanghai,2024-09-24T06:15:30.000Z,30,null,",
4392+
"2024-09-24T06:15:31.000Z,shanghai,2024-09-24T06:15:31.000Z,null,31000,",
4393+
"2024-09-24T06:15:35.000Z,shanghai,2024-09-24T06:15:35.000Z,null,35000,",
4394+
"2024-09-24T06:15:36.000Z,shanghai,2024-09-24T06:15:36.000Z,36,null,",
4395+
"2024-09-24T06:15:40.000Z,shanghai,2024-09-24T06:15:40.000Z,40,40000,",
4396+
"2024-09-24T06:15:41.000Z,shanghai,2024-09-24T06:15:41.000Z,41,null,",
4397+
"2024-09-24T06:15:46.000Z,shanghai,2024-09-24T06:15:46.000Z,null,46000,",
4398+
"2024-09-24T06:15:50.000Z,shanghai,2024-09-24T06:15:50.000Z,null,50000,",
4399+
"2024-09-24T06:15:51.000Z,shanghai,2024-09-24T06:15:51.000Z,null,null,",
4400+
"2024-09-24T06:15:55.000Z,shanghai,2024-09-24T06:15:55.000Z,55,null,",
4401+
},
4402+
DATABASE_NAME);
4403+
}
4404+
43694405
@Test
43704406
public void exceptionTest() {
43714407
tableAssertTestFail(
@@ -4456,6 +4492,22 @@ public void exceptionTest() {
44564492
"select 1 as g, approx_percentile(s1,s2,0.5) from table1 group by 1",
44574493
"701: Aggregation functions [approx_percentile] do not support weight as INT64 type",
44584494
DATABASE_NAME);
4495+
tableAssertTestFail(
4496+
"select percentile() from table1",
4497+
"701: Aggregation functions [percentile] should only have two arguments",
4498+
DATABASE_NAME);
4499+
tableAssertTestFail(
4500+
"select percentile(s1,1.1) from table1",
4501+
"701: percentage should be in [0,1], got 1.1",
4502+
DATABASE_NAME);
4503+
tableAssertTestFail(
4504+
"select percentile(s1,'test') from table1",
4505+
"701: The second argument of 'percentile' function percentage must be a double literal",
4506+
DATABASE_NAME);
4507+
tableAssertTestFail(
4508+
"select percentile(s5,0.5) from table1",
4509+
"701: Aggregation functions [percentile] should have value column as numeric type [INT32, INT64, FLOAT, DOUBLE, TIMESTAMP]",
4510+
DATABASE_NAME);
44594511
}
44604512

44614513
// ==================================================================

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

Lines changed: 159 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,159 @@
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+
for (int i = 0; i < size; i++) {
133+
ReadWriteIOUtils.write(values[i], buffer);
134+
}
135+
}
136+
137+
public static Percentile deserialize(ByteBuffer buffer) {
138+
int size = ReadWriteIOUtils.readInt(buffer);
139+
Percentile percentile = new Percentile();
140+
if (size > percentile.capacity) {
141+
percentile.capacity = size;
142+
percentile.values = new double[size];
143+
}
144+
percentile.size = size;
145+
for (int i = 0; i < size; i++) {
146+
percentile.values[i] = ReadWriteIOUtils.readDouble(buffer);
147+
}
148+
return percentile;
149+
}
150+
151+
public int getSerializedSize() {
152+
return Integer.BYTES + (size * Double.BYTES);
153+
}
154+
155+
public long getEstimatedSize() {
156+
return RamUsageEstimator.shallowSizeOfInstance(Percentile.class)
157+
+ (long) capacity * Double.BYTES;
158+
}
159+
}

‎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;
@@ -290,6 +291,8 @@ private static GroupedAccumulator createBuiltinGroupedAccumulator(
290291
} else {
291292
return new GroupedApproxPercentileWithWeightAccumulator(inputDataTypes.get(0));
292293
}
294+
case PERCENTILE:
295+
return new GroupedPercentileAccumulator(inputDataTypes.get(0));
293296
default:
294297
throw new IllegalArgumentException("Invalid Aggregation function: " + aggregationType);
295298
}
@@ -353,6 +356,8 @@ public static TableAccumulator createBuiltinAccumulator(
353356
} else {
354357
return new ApproxPercentileWithWeightAccumulator(inputDataTypes.get(0));
355358
}
359+
case PERCENTILE:
360+
return new PercentileAccumulator(inputDataTypes.get(0));
356361
default:
357362
throw new IllegalArgumentException("Invalid Aggregation function: " + aggregationType);
358363
}

0 commit comments

Comments
 (0)