Skip to content

Commit ce8a0ef

Browse files
FearfulTomcat27FearfulTomcat27
committed
feat: Implement GroupedPercentileAccumulator and PercentileAccumulator for percentile aggregation
1 parent 07d3618 commit ce8a0ef

6 files changed

Lines changed: 562 additions & 5 deletions

File tree

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/source/relational/Percentile.java renamed to iotdb-core/calc-commons/src/main/java/org/apache/iotdb/calc/execution/operator/source/relational/Percentile.java

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -12,9 +12,9 @@
1212
* limitations under the License.
1313
*/
1414

15-
package org.apache.iotdb.db.queryengine.execution.operator.source.relational;
15+
package org.apache.iotdb.calc.execution.operator.source.relational;
1616

17-
import org.apache.iotdb.db.exception.sql.SemanticException;
17+
import org.apache.iotdb.commons.exception.SemanticException;
1818

1919
import org.apache.tsfile.utils.RamUsageEstimator;
2020
import org.apache.tsfile.utils.ReadWriteIOUtils;

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

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -43,6 +43,7 @@
4343
import org.apache.iotdb.calc.execution.operator.source.relational.aggregation.grouped.GroupedMinAccumulator;
4444
import org.apache.iotdb.calc.execution.operator.source.relational.aggregation.grouped.GroupedMinByAccumulator;
4545
import org.apache.iotdb.calc.execution.operator.source.relational.aggregation.grouped.GroupedModeAccumulator;
46+
import org.apache.iotdb.calc.execution.operator.source.relational.aggregation.grouped.GroupedPercentileAccumulator;
4647
import org.apache.iotdb.calc.execution.operator.source.relational.aggregation.grouped.GroupedSumAccumulator;
4748
import org.apache.iotdb.calc.execution.operator.source.relational.aggregation.grouped.GroupedUserDefinedAggregateAccumulator;
4849
import org.apache.iotdb.calc.execution.operator.source.relational.aggregation.grouped.GroupedVarianceAccumulator;

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

Lines changed: 226 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,226 @@
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.calc.execution.operator.source.relational.aggregation;
16+
17+
import org.apache.iotdb.calc.execution.operator.source.relational.Percentile;
18+
import org.apache.iotdb.commons.exception.SemanticException;
19+
20+
import org.apache.tsfile.block.column.Column;
21+
import org.apache.tsfile.block.column.ColumnBuilder;
22+
import org.apache.tsfile.enums.TSDataType;
23+
import org.apache.tsfile.file.metadata.statistics.Statistics;
24+
import org.apache.tsfile.utils.Binary;
25+
import org.apache.tsfile.utils.RamUsageEstimator;
26+
import org.apache.tsfile.utils.ReadWriteIOUtils;
27+
import org.apache.tsfile.write.UnSupportedDataTypeException;
28+
29+
import java.nio.ByteBuffer;
30+
31+
public class PercentileAccumulator implements TableAccumulator {
32+
private static final long INSTANCE_SIZE =
33+
RamUsageEstimator.shallowSizeOfInstance(PercentileAccumulator.class);
34+
35+
private final TSDataType seriesDataType;
36+
private Percentile percentile = new Percentile();
37+
private double percentage;
38+
39+
public PercentileAccumulator(TSDataType seriesDataType) {
40+
this.seriesDataType = seriesDataType;
41+
}
42+
43+
@Override
44+
public long getEstimatedSize() {
45+
return INSTANCE_SIZE + percentile.getEstimatedSize();
46+
}
47+
48+
@Override
49+
public TableAccumulator copy() {
50+
return new PercentileAccumulator(seriesDataType);
51+
}
52+
53+
@Override
54+
public void addInput(Column[] arguments, AggregationMask mask) {
55+
if (arguments.length != 2) {
56+
throw new SemanticException(
57+
String.format("PERCENTILE requires 2 arguments, but got %d", arguments.length));
58+
}
59+
percentage = arguments[1].getDouble(0);
60+
switch (seriesDataType) {
61+
case INT32:
62+
addIntInput(arguments[0], mask);
63+
return;
64+
case INT64:
65+
case TIMESTAMP:
66+
addLongInput(arguments[0], mask);
67+
return;
68+
case FLOAT:
69+
addFloatInput(arguments[0], mask);
70+
return;
71+
case DOUBLE:
72+
addDoubleInput(arguments[0], mask);
73+
return;
74+
default:
75+
throw new UnSupportedDataTypeException(
76+
String.format("Unsupported data type in Percentile Aggregation: %s", seriesDataType));
77+
}
78+
}
79+
80+
@Override
81+
public void addIntermediate(Column argument) {
82+
for (int i = 0; i < argument.getPositionCount(); i++) {
83+
if (!argument.isNull(i)) {
84+
byte[] data = argument.getBinary(i).getValues();
85+
ByteBuffer buffer = ByteBuffer.wrap(data);
86+
this.percentage = ReadWriteIOUtils.readDouble(buffer);
87+
percentile.merge(Percentile.deserialize(buffer));
88+
}
89+
}
90+
}
91+
92+
@Override
93+
public void evaluateIntermediate(ColumnBuilder columnBuilder) {
94+
int percentileDataLength = percentile.getSerializedSize();
95+
ByteBuffer buffer = ByteBuffer.allocate(8 + percentileDataLength);
96+
ReadWriteIOUtils.write(percentage, buffer);
97+
percentile.serialize(buffer);
98+
columnBuilder.writeBinary(new Binary(buffer.array()));
99+
}
100+
101+
@Override
102+
public void evaluateFinal(ColumnBuilder columnBuilder) {
103+
double result = percentile.getPercentile(percentage);
104+
if (Double.isNaN(result)) {
105+
columnBuilder.appendNull();
106+
return;
107+
}
108+
switch (seriesDataType) {
109+
case INT32:
110+
columnBuilder.writeInt((int) result);
111+
break;
112+
case INT64:
113+
case TIMESTAMP:
114+
columnBuilder.writeLong((long) result);
115+
break;
116+
case FLOAT:
117+
columnBuilder.writeFloat((float) result);
118+
break;
119+
case DOUBLE:
120+
columnBuilder.writeDouble(result);
121+
break;
122+
default:
123+
throw new UnSupportedDataTypeException(
124+
String.format("Unsupported data type in PERCENTILE Aggregation: %s", seriesDataType));
125+
}
126+
}
127+
128+
@Override
129+
public boolean hasFinalResult() {
130+
return false;
131+
}
132+
133+
@Override
134+
public void addStatistics(Statistics[] statistics) {
135+
throw new UnsupportedOperationException("PercentileAccumulator does not support statistics");
136+
}
137+
138+
@Override
139+
public void reset() {
140+
percentile.clear();
141+
}
142+
143+
private void addIntInput(Column column, AggregationMask mask) {
144+
int positionCount = mask.getSelectedPositionCount();
145+
146+
if (mask.isSelectAll()) {
147+
for (int i = 0; i < positionCount; i++) {
148+
if (!column.isNull(i)) {
149+
percentile.addValue(column.getInt(i));
150+
}
151+
}
152+
} else {
153+
int[] selectedPositions = mask.getSelectedPositions();
154+
int position;
155+
for (int i = 0; i < positionCount; i++) {
156+
position = selectedPositions[i];
157+
if (!column.isNull(position)) {
158+
percentile.addValue(column.getInt(position));
159+
}
160+
}
161+
}
162+
}
163+
164+
private void addLongInput(Column column, AggregationMask mask) {
165+
int positionCount = mask.getSelectedPositionCount();
166+
167+
if (mask.isSelectAll()) {
168+
for (int i = 0; i < positionCount; i++) {
169+
if (!column.isNull(i)) {
170+
percentile.addValue(column.getLong(i));
171+
}
172+
}
173+
} else {
174+
int[] selectedPositions = mask.getSelectedPositions();
175+
int position;
176+
for (int i = 0; i < positionCount; i++) {
177+
position = selectedPositions[i];
178+
if (!column.isNull(position)) {
179+
percentile.addValue(column.getLong(position));
180+
}
181+
}
182+
}
183+
}
184+
185+
private void addFloatInput(Column column, AggregationMask mask) {
186+
int positionCount = mask.getSelectedPositionCount();
187+
188+
if (mask.isSelectAll()) {
189+
for (int i = 0; i < positionCount; i++) {
190+
if (!column.isNull(i)) {
191+
percentile.addValue(column.getFloat(i));
192+
}
193+
}
194+
} else {
195+
int[] selectedPositions = mask.getSelectedPositions();
196+
int position;
197+
for (int i = 0; i < positionCount; i++) {
198+
position = selectedPositions[i];
199+
if (!column.isNull(position)) {
200+
percentile.addValue(column.getFloat(position));
201+
}
202+
}
203+
}
204+
}
205+
206+
private void addDoubleInput(Column column, AggregationMask mask) {
207+
int positionCount = mask.getSelectedPositionCount();
208+
209+
if (mask.isSelectAll()) {
210+
for (int i = 0; i < positionCount; i++) {
211+
if (!column.isNull(i)) {
212+
percentile.addValue(column.getDouble(i));
213+
}
214+
}
215+
} else {
216+
int[] selectedPositions = mask.getSelectedPositions();
217+
int position;
218+
for (int i = 0; i < positionCount; i++) {
219+
position = selectedPositions[i];
220+
if (!column.isNull(position)) {
221+
percentile.addValue(column.getDouble(position));
222+
}
223+
}
224+
}
225+
}
226+
}

0 commit comments

Comments
 (0)