Skip to content

Commit 59823b0

Browse files
FearfulTomcat27FearfulTomcat27
committed
fix: Refactor APPROX_PERCENTILE accumulators for improved argument handling and validation
1 parent a73efb0 commit 59823b0

8 files changed

Lines changed: 41 additions & 123 deletions

File tree

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

Lines changed: 1 addition & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -19,7 +19,6 @@
1919

2020
package org.apache.iotdb.relational.it.query.recent;
2121

22-
import org.apache.iotdb.db.queryengine.execution.operator.source.relational.aggregation.approximate.TDigest;
2322
import org.apache.iotdb.it.env.EnvFactory;
2423
import org.apache.iotdb.it.framework.IoTDBTestRunner;
2524
import org.apache.iotdb.itbase.category.TableClusterIT;
@@ -4326,10 +4325,6 @@ public void approxPercentileTest() {
43264325
"2024-09-24T06:15:55.000Z,shanghai,55,0,",
43274326
},
43284327
DATABASE_NAME);
4329-
4330-
TDigest a = new TDigest();
4331-
TDigest b = new TDigest();
4332-
a.add(b);
43334328
}
43344329

43354330
@Test
@@ -4400,7 +4395,7 @@ public void exceptionTest() {
44004395
DATABASE_NAME);
44014396
tableAssertTestFail(
44024397
"select approx_percentile(s1,1.1) from table1",
4403-
"701: q should be in [0,1], got 1.1",
4398+
"701: percentage should be in [0,1], got 1.1",
44044399
DATABASE_NAME);
44054400
tableAssertTestFail(
44064401
"select approx_percentile(s1,'test') from table1",

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

Lines changed: 5 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -36,7 +36,7 @@ public abstract class AbstractApproxPercentileAccumulator implements TableAccumu
3636
protected final TSDataType seriesDataType;
3737
protected double percentage;
3838

39-
public AbstractApproxPercentileAccumulator(TSDataType seriesDataType) {
39+
AbstractApproxPercentileAccumulator(TSDataType seriesDataType) {
4040
this.seriesDataType = seriesDataType;
4141
}
4242

@@ -90,21 +90,19 @@ public void addIntermediate(Column argument) {
9090
// Read percentage from the first 8 bytes and TDigest from the rest
9191
ByteBuffer buffer = ByteBuffer.wrap(data);
9292
this.percentage = ReadWriteIOUtils.readDouble(buffer);
93-
byte[] tDigestData = new byte[data.length - 8];
94-
buffer.get(tDigestData);
95-
TDigest other = TDigest.fromByteArray(tDigestData);
93+
TDigest other = TDigest.fromByteBuffer(buffer);
9694
tDigest.add(other);
9795
}
9896
}
9997
}
10098

10199
@Override
102100
public void evaluateIntermediate(ColumnBuilder columnBuilder) {
103-
byte[] tDigestData = tDigest.toByteArray();
101+
int tDigestDataLength = tDigest.byteSize();
104102
// Create a buffer with space for percentage (8 bytes) + TDigest data
105-
ByteBuffer buffer = ByteBuffer.allocate(8 + tDigestData.length);
103+
ByteBuffer buffer = ByteBuffer.allocate(8 + tDigestDataLength);
106104
ReadWriteIOUtils.write(percentage, buffer);
107-
buffer.put(tDigestData);
105+
tDigest.toByteArray(buffer);
108106
columnBuilder.writeBinary(new Binary(buffer.array()));
109107
}
110108

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

Lines changed: 13 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -58,9 +58,9 @@ public class TDigest implements Serializable {
5858
private double max = Double.NEGATIVE_INFINITY;
5959

6060
// Configuration flags
61-
public boolean useAlternatingSort = true;
62-
public boolean useTwoLevelCompression = true;
63-
public static boolean useWeightLimit = true;
61+
private boolean useAlternatingSort = true;
62+
private boolean useTwoLevelCompression = true;
63+
private boolean useWeightLimit = true;
6464

6565
public TDigest() {
6666
this(DEFAULT_COMPRESSION);
@@ -514,7 +514,7 @@ public double cdf(double x) {
514514
*/
515515
public double quantile(double q) {
516516
if (q < 0 || q > 1) {
517-
throw new SemanticException("q should be in [0,1], got " + q);
517+
throw new SemanticException("percentage should be in [0,1], got " + q);
518518
}
519519
mergeNewValues();
520520

@@ -745,6 +745,11 @@ public byte[] toByteArray() {
745745
return buf.array();
746746
}
747747

748+
public byte[] toByteArray(ByteBuffer buffer) {
749+
asBytes(buffer);
750+
return buffer.array();
751+
}
752+
748753
/**
749754
* Serialize to byte array using small encoding.
750755
*
@@ -766,6 +771,10 @@ public static TDigest fromByteArray(byte[] bytes) {
766771
return fromBytes(ByteBuffer.wrap(bytes));
767772
}
768773

774+
public static TDigest fromByteBuffer(ByteBuffer buffer) {
775+
return fromBytes(buffer);
776+
}
777+
769778
/** Over-ride the min and max values (used internally during deserialization). */
770779
private void setMinMax(double min, double max) {
771780
this.min = min;

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

Lines changed: 0 additions & 82 deletions
This file was deleted.

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

Lines changed: 8 additions & 10 deletions
Original file line numberDiff line numberDiff line change
@@ -14,7 +14,6 @@
1414

1515
package org.apache.iotdb.db.queryengine.execution.operator.source.relational.aggregation.grouped;
1616

17-
import org.apache.iotdb.db.exception.sql.SemanticException;
1817
import org.apache.iotdb.db.queryengine.execution.operator.source.relational.aggregation.AggregationMask;
1918
import org.apache.iotdb.db.queryengine.execution.operator.source.relational.aggregation.approximate.TDigest;
2019
import org.apache.iotdb.db.queryengine.execution.operator.source.relational.aggregation.grouped.array.TDigestBigArray;
@@ -37,7 +36,7 @@ public abstract class AbstractGroupedApproxPercentileAccumulator implements Grou
3736
protected double percentage;
3837
protected final TDigestBigArray array = new TDigestBigArray();
3938

40-
public AbstractGroupedApproxPercentileAccumulator(TSDataType seriesDataType) {
39+
AbstractGroupedApproxPercentileAccumulator(TSDataType seriesDataType) {
4140
this.seriesDataType = seriesDataType;
4241
}
4342

@@ -58,7 +57,7 @@ public void addInput(int[] groupIds, Column[] arguments, AggregationMask mask) {
5857
} else if (arguments.length == 3) {
5958
percentage = arguments[2].getDouble(0);
6059
} else {
61-
throw new SemanticException(
60+
throw new IllegalArgumentException(
6261
String.format(
6362
"APPROX_PERCENTILE requires 2 or 3 arguments, but got %d", arguments.length));
6463
}
@@ -92,20 +91,19 @@ public void addIntermediate(int[] groupIds, Column argument) {
9291
byte[] data = argument.getBinary(i).getValues();
9392
ByteBuffer buffer = ByteBuffer.wrap(data);
9493
this.percentage = ReadWriteIOUtils.readDouble(buffer);
95-
byte[] tDigestData = new byte[data.length - 8];
96-
buffer.get(tDigestData);
97-
TDigest other = TDigest.fromByteArray(tDigestData);
94+
TDigest other = TDigest.fromByteBuffer(buffer);
9895
array.get(groupId).add(other);
9996
}
10097
}
10198
}
10299

103100
@Override
104101
public void evaluateIntermediate(int groupId, ColumnBuilder columnBuilder) {
105-
byte[] tDigestData = array.get(groupId).toByteArray();
106-
ByteBuffer buffer = ByteBuffer.allocate(8 + tDigestData.length);
102+
TDigest tDigest = array.get(groupId);
103+
int tDigestDataLength = tDigest.byteSize();
104+
ByteBuffer buffer = ByteBuffer.allocate(8 + tDigestDataLength);
107105
ReadWriteIOUtils.write(percentage, buffer);
108-
buffer.put(tDigestData);
106+
tDigest.toByteArray(buffer);
109107
columnBuilder.writeBinary(new Binary(buffer.array()));
110108
}
111109

@@ -129,7 +127,7 @@ public void evaluateFinal(int groupId, ColumnBuilder columnBuilder) {
129127
default:
130128
throw new UnSupportedDataTypeException(
131129
String.format(
132-
"Unsupported data type in APPROX_COUNT_DISTINCT Aggregation: %s", seriesDataType));
130+
"Unsupported data type in APPROX_PERCENTILE Aggregation: %s", seriesDataType));
133131
}
134132
}
135133

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

Lines changed: 6 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -143,7 +143,7 @@ public void addBooleanInput(
143143
groupId = groupIds[position];
144144
HyperLogLog hll = hlls.get(groupId, maxStandardError);
145145
if (!column.isNull(position)) {
146-
hll.add(column.getBoolean(i));
146+
hll.add(column.getBoolean(position));
147147
}
148148
}
149149
}
@@ -174,7 +174,7 @@ public void addIntInput(
174174
groupId = groupIds[position];
175175
HyperLogLog hll = hlls.get(groupId, maxStandardError);
176176
if (!column.isNull(position)) {
177-
hll.add(column.getInt(i));
177+
hll.add(column.getInt(position));
178178
}
179179
}
180180
}
@@ -205,7 +205,7 @@ public void addLongInput(
205205
groupId = groupIds[position];
206206
HyperLogLog hll = hlls.get(groupId, maxStandardError);
207207
if (!column.isNull(position)) {
208-
hll.add(column.getLong(i));
208+
hll.add(column.getLong(position));
209209
}
210210
}
211211
}
@@ -236,7 +236,7 @@ public void addFloatInput(
236236
groupId = groupIds[position];
237237
HyperLogLog hll = hlls.get(groupId, maxStandardError);
238238
if (!column.isNull(position)) {
239-
hll.add(column.getFloat(i));
239+
hll.add(column.getFloat(position));
240240
}
241241
}
242242
}
@@ -267,7 +267,7 @@ public void addDoubleInput(
267267
groupId = groupIds[position];
268268
HyperLogLog hll = hlls.get(groupId, maxStandardError);
269269
if (!column.isNull(position)) {
270-
hll.add(column.getDouble(i));
270+
hll.add(column.getDouble(position));
271271
}
272272
}
273273
}
@@ -298,7 +298,7 @@ public void addBinaryInput(
298298
groupId = groupIds[position];
299299
HyperLogLog hll = hlls.get(groupId, maxStandardError);
300300
if (!column.isNull(position)) {
301-
hll.add(column.getBinary(i));
301+
hll.add(column.getBinary(position));
302302
}
303303
}
304304
}

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

Lines changed: 4 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -49,7 +49,7 @@ public void addIntInput(int[] groupIds, Column[] arguments, AggregationMask mask
4949
groupId = groupIds[position];
5050
TDigest tDigest = array.get(groupId);
5151
if (!valueColumn.isNull(position)) {
52-
tDigest.add(valueColumn.getInt(i));
52+
tDigest.add(valueColumn.getInt(position));
5353
}
5454
}
5555
}
@@ -78,7 +78,7 @@ public void addLongInput(int[] groupIds, Column[] arguments, AggregationMask mas
7878
groupId = groupIds[position];
7979
TDigest tDigest = array.get(groupId);
8080
if (!valueColumn.isNull(position)) {
81-
tDigest.add(valueColumn.getLong(i));
81+
tDigest.add(valueColumn.getLong(position));
8282
}
8383
}
8484
}
@@ -107,7 +107,7 @@ public void addFloatInput(int[] groupIds, Column[] arguments, AggregationMask ma
107107
groupId = groupIds[position];
108108
TDigest tDigest = array.get(groupId);
109109
if (!valueColumn.isNull(position)) {
110-
tDigest.add(valueColumn.getFloat(i));
110+
tDigest.add(valueColumn.getFloat(position));
111111
}
112112
}
113113
}
@@ -136,7 +136,7 @@ public void addDoubleInput(int[] groupIds, Column[] arguments, AggregationMask m
136136
groupId = groupIds[position];
137137
TDigest tDigest = array.get(groupId);
138138
if (!valueColumn.isNull(position)) {
139-
tDigest.add(valueColumn.getDouble(i));
139+
tDigest.add(valueColumn.getDouble(position));
140140
}
141141
}
142142
}

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

Lines changed: 4 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -51,7 +51,7 @@ public void addIntInput(int[] groupIds, Column[] arguments, AggregationMask mask
5151
groupId = groupIds[position];
5252
TDigest tDigest = array.get(groupId);
5353
if (!valueColumn.isNull(position)) {
54-
tDigest.add(valueColumn.getInt(i), weightColumn.getInt(i));
54+
tDigest.add(valueColumn.getInt(position), weightColumn.getInt(position));
5555
}
5656
}
5757
}
@@ -81,7 +81,7 @@ public void addLongInput(int[] groupIds, Column[] arguments, AggregationMask mas
8181
groupId = groupIds[position];
8282
TDigest tDigest = array.get(groupId);
8383
if (!valueColumn.isNull(position)) {
84-
tDigest.add(valueColumn.getLong(i), weightColumn.getInt(i));
84+
tDigest.add(valueColumn.getLong(position), weightColumn.getInt(position));
8585
}
8686
}
8787
}
@@ -111,7 +111,7 @@ public void addFloatInput(int[] groupIds, Column[] arguments, AggregationMask ma
111111
groupId = groupIds[position];
112112
TDigest tDigest = array.get(groupId);
113113
if (!valueColumn.isNull(position)) {
114-
tDigest.add(valueColumn.getFloat(i), weightColumn.getInt(i));
114+
tDigest.add(valueColumn.getFloat(position), weightColumn.getInt(position));
115115
}
116116
}
117117
}
@@ -141,7 +141,7 @@ public void addDoubleInput(int[] groupIds, Column[] arguments, AggregationMask m
141141
groupId = groupIds[position];
142142
TDigest tDigest = array.get(groupId);
143143
if (!valueColumn.isNull(position)) {
144-
tDigest.add(valueColumn.getDouble(i), weightColumn.getInt(i));
144+
tDigest.add(valueColumn.getDouble(position), weightColumn.getInt(position));
145145
}
146146
}
147147
}

0 commit comments

Comments
 (0)