Add PERCENTILE aggregation function and related validation #16545
There are no files selected for viewing
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,173 @@ | ||
| /* | ||
| * Licensed to the Apache Software Foundation (ASF) under one | ||
| * or more contributor license agreements. See the NOTICE file | ||
| * distributed with this work for additional information | ||
| * regarding copyright ownership. The ASF licenses this file | ||
| * to you under the Apache License, Version 2.0 (the | ||
| * "License"); you may not use this file except in compliance | ||
| * with the License. You may obtain a copy of the License at | ||
| * | ||
| * http://www.apache.org/licenses/LICENSE-2.0 | ||
| * | ||
| * Unless required by applicable law or agreed to in writing, | ||
| * software distributed under the License is distributed on an | ||
| * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY | ||
| * KIND, either express or implied. See the License for the | ||
| * specific language governing permissions and limitations | ||
| * under the License. | ||
| */ | ||
|
|
||
| package org.apache.iotdb.calc.execution.operator.source.relational; | ||
|
|
||
| import org.apache.iotdb.commons.exception.SemanticException; | ||
|
|
||
| import org.apache.tsfile.utils.RamUsageEstimator; | ||
| import org.apache.tsfile.utils.ReadWriteIOUtils; | ||
|
|
||
| import java.nio.ByteBuffer; | ||
| import java.util.Arrays; | ||
|
|
||
| public class Percentile { | ||
| private double[] values; | ||
| private int size; | ||
| private int capacity; | ||
| private boolean sorted; | ||
|
|
||
| private static final int INITIAL_CAPACITY = 32; | ||
| private static final double GROWTH_FACTOR = 1.5; | ||
|
|
||
| public Percentile() { | ||
| this.capacity = INITIAL_CAPACITY; | ||
| this.values = new double[capacity]; | ||
| this.size = 0; | ||
| this.sorted = true; | ||
| } | ||
|
|
||
| public void addValue(double value) { | ||
| ensureCapacity(); | ||
| values[size++] = value; | ||
| sorted = false; | ||
| } | ||
|
|
||
| public void addValues(double... vals) { | ||
| if (vals == null || vals.length == 0) { | ||
| return; | ||
| } | ||
|
|
||
| int newSize = size + vals.length; | ||
| if (newSize > capacity) { | ||
| grow(newSize); | ||
| } | ||
|
|
||
| System.arraycopy(vals, 0, values, size, vals.length); | ||
| size = newSize; | ||
| sorted = false; | ||
| } | ||
|
|
||
| public void merge(Percentile other) { | ||
| if (other == null || other.size == 0) { | ||
| return; | ||
| } | ||
|
|
||
| int newSize = size + other.size; | ||
| if (newSize > capacity) { | ||
| grow(newSize); | ||
| } | ||
|
|
||
| System.arraycopy(other.values, 0, values, size, other.size); | ||
| size = newSize; | ||
| sorted = false; | ||
| } | ||
|
|
||
| public double getPercentile(double percentile) { | ||
| if (size == 0) { | ||
| return Double.NaN; | ||
| } | ||
| if (percentile < 0.0 || percentile > 1.0) { | ||
| throw new SemanticException("percentage should be in [0,1], got " + percentile); | ||
| } | ||
|
|
||
| ensureSorted(); | ||
|
|
||
| if (size == 1) { | ||
| return values[0]; | ||
| } | ||
|
|
||
| double realIndex = percentile * (size - 1); | ||
| int index = (int) realIndex; | ||
| double fraction = realIndex - index; | ||
|
|
||
| if (index >= size - 1) { | ||
| return values[size - 1]; | ||
| } | ||
|
|
||
| return values[index] + fraction * (values[index + 1] - values[index]); | ||
| } | ||
|
|
||
| public int getSize() { | ||
| return size; | ||
| } | ||
|
|
||
| public void clear() { | ||
| // Shrink the backing array back to the initial capacity so the memory held by a large group is | ||
| // actually released on reset, instead of staying reserved at the historical peak capacity. | ||
| if (capacity > INITIAL_CAPACITY) { | ||
| capacity = INITIAL_CAPACITY; | ||
| values = new double[capacity]; | ||
| } | ||
| size = 0; | ||
| sorted = true; | ||
| } | ||
|
|
||
| private void ensureCapacity() { | ||
| if (size >= capacity) { | ||
| grow(size + 1); | ||
| } | ||
| } | ||
|
|
||
| private void grow(int minCapacity) { | ||
| int newCapacity = Math.max((int) (capacity * GROWTH_FACTOR), minCapacity); | ||
| double[] newValues = new double[newCapacity]; | ||
| System.arraycopy(values, 0, newValues, 0, size); | ||
| values = newValues; | ||
| capacity = newCapacity; | ||
| } | ||
|
|
||
| private void ensureSorted() { | ||
| if (!sorted && size > 1) { | ||
| Arrays.sort(values, 0, size); | ||
| sorted = true; | ||
| } | ||
| } | ||
|
|
||
| public void serialize(ByteBuffer buffer) { | ||
| ReadWriteIOUtils.write(size, buffer); | ||
| for (int i = 0; i < size; i++) { | ||
| ReadWriteIOUtils.write(values[i], buffer); | ||
| } | ||
| } | ||
|
|
||
| public static Percentile deserialize(ByteBuffer buffer) { | ||
|
Contributor
There was a problem hiding this comment. 🟡 [健壮性 / 隐患] 当前尚未真正触发 bug，因为反序列化结果只通过 建议在此显式设置 |
||
| int size = ReadWriteIOUtils.readInt(buffer); | ||
| Percentile percentile = new Percentile(); | ||
| if (size > percentile.capacity) { | ||
| percentile.capacity = size; | ||
| percentile.values = new double[size]; | ||
| } | ||
|
Comment on lines
+152
to
+156
There was a problem hiding this comment. Percentile.deserialize() does not restore the 'sorted' state. Since serialize() writes values in their current internal order (which may be unsorted), leaving the deserialized instance with sorted=true can lead to incorrect getPercentile() results if the deserialized object is queried directly. Set sorted=false on deserialization (or serialize in sorted order) to preserve correctness. |
||
| percentile.size = size; | ||
| for (int i = 0; i < size; i++) { | ||
| percentile.values[i] = ReadWriteIOUtils.readDouble(buffer); | ||
| } | ||
| percentile.sorted = false; | ||
| return percentile; | ||
| } | ||
|
|
||
| public int getSerializedSize() { | ||
|
Contributor
There was a problem hiding this comment. 🔵 [可扩展性 / 溢出] 需要意识到
建议：至少把 |
||
| return Integer.BYTES + (int) ((long) size * Double.BYTES); | ||
| } | ||
|
|
||
| public long getEstimatedSize() { | ||
| return RamUsageEstimator.shallowSizeOfInstance(Percentile.class) | ||
| + (long) capacity * Double.BYTES; | ||
| } | ||
| } | ||
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
🟡 [License 头 / CI 风险] 本仓库 Java 文件统一使用 ASF 长版权头（
Licensed to the Apache Software Foundation (ASF) under one ...；本包下现有 112/112 个文件均如此），而本 PR 新增文件用的是简化版 Apache 头。仓库通过license-maven-plugin校验头部（见根pom.xml），这很可能导致 License check CI 失败。请将本 PR 所有新增文件的头替换为长版 ASF 头：
Percentile.java、PercentileAccumulator.java、GroupedPercentileAccumulator.java、PercentileBigArray.java。直接复制同目录现有文件的头即可。