JackieTien97 commented on code in PR #16545: URL: https://github.com/apache/iotdb/pull/16545#discussion_r3366851243
########## iotdb-core/calc-commons/src/main/java/org/apache/iotdb/calc/execution/operator/source/relational/aggregation/grouped/array/PercentileBigArray.java: ########## @@ -0,0 +1,76 @@ +/* + * Licensed 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.aggregation.grouped.array; + +import org.apache.iotdb.calc.execution.operator.source.relational.Percentile; + +import static org.apache.tsfile.utils.RamUsageEstimator.shallowSizeOf; +import static org.apache.tsfile.utils.RamUsageEstimator.shallowSizeOfInstance; + +public final class PercentileBigArray { + private static final long INSTANCE_SIZE = shallowSizeOfInstance(PercentileBigArray.class); + private final ObjectBigArray<Percentile> array; + private long sizeOfPercentile; + + public PercentileBigArray() { + array = new ObjectBigArray<>(); + } + + public long sizeOf() { Review Comment: 🟠 **[内存核算]** `sizeOf()` 返回值会被严重低估，使本 PR 引入 `MemoryReservationManager` 的初衷基本失效。 `sizeOfPercentile` 只在 `updateRetainedSize()` 中更新，而它仅在 `get()` 首次为某分组创建 `Percentile` 时（经 `set()`）被调用一次，统计的是初始容量（32 个 double）。之后 accumulator 通过 `get(groupId).addValue(...)` / `merge(...)` 不断向 `Percentile` 追加并扩容，但 `set()`/`updateRetainedSize()` 不会再被调用，**扩容后的真实占用从未被计入**。 于是 `GroupedPercentileAccumulator.getEstimatedSize()` → `updateMemoryReservation()` 预留的内存远小于实际，大分组下既不会触发反压也无法防止 OOM。对照非分组版 `PercentileAccumulator` 是用 `percentile.getEstimatedSize()` 实时计算的，无此问题。 建议：在 `sizeOf()` 时实时累加各分组 `Percentile.getEstimatedSize()`，或在每次 `addValue`/`merge` 后同步 `sizeOfPercentile`。 ########## iotdb-core/calc-commons/src/main/java/org/apache/iotdb/calc/execution/operator/source/relational/Percentile.java: ########## @@ -0,0 +1,161 @@ +/* + * Licensed 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() { + 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) { + int size = ReadWriteIOUtils.readInt(buffer); + Percentile percentile = new Percentile(); + if (size > percentile.capacity) { + percentile.capacity = size; + percentile.values = new double[size]; + } + percentile.size = size; + for (int i = 0; i < size; i++) { + percentile.values[i] = ReadWriteIOUtils.readDouble(buffer); + } + return percentile; + } + + public int getSerializedSize() { Review Comment: 🔵 **[可扩展性 / 溢出]** 需要意识到 `percentile` 是**精确**实现，会在内存中保留分组内全部原始值，并在两阶段聚合时把所有值序列化在节点间传输——这与 `approx_percentile`(TDigest, 常数空间) 是根本不同的取舍。 `getSerializedSize()` 用 `size * Double.BYTES`（int 运算），当单组 `size` 超过约 2.68 亿时整型溢出得到负值；`ByteBuffer.allocate()` 也有 2GB 上限；`grow()` 里 `(int)(capacity * 1.5)` 同样可能溢出。大基数场景既是 OOM 风险也是溢出风险。 建议：至少把 `size * Double.BYTES` 改为 `(long) size * Double.BYTES` 避免静默溢出；并考虑对单组规模设上限并给出明确报错，文档中说明与 `approx_percentile` 的取舍。 ########## iotdb-core/calc-commons/src/main/java/org/apache/iotdb/calc/execution/operator/source/relational/aggregation/grouped/GroupedPercentileAccumulator.java: ########## @@ -0,0 +1,272 @@ +/* + * Licensed 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.aggregation.grouped; + +import org.apache.iotdb.calc.execution.operator.source.relational.Percentile; +import org.apache.iotdb.calc.execution.operator.source.relational.aggregation.AggregationMask; +import org.apache.iotdb.calc.execution.operator.source.relational.aggregation.grouped.array.PercentileBigArray; +import org.apache.iotdb.calc.plan.planner.memory.MemoryReservationManager; +import org.apache.iotdb.commons.exception.SemanticException; + +import org.apache.tsfile.block.column.Column; +import org.apache.tsfile.block.column.ColumnBuilder; +import org.apache.tsfile.enums.TSDataType; +import org.apache.tsfile.utils.Binary; +import org.apache.tsfile.utils.RamUsageEstimator; +import org.apache.tsfile.utils.ReadWriteIOUtils; +import org.apache.tsfile.write.UnSupportedDataTypeException; + +import java.nio.ByteBuffer; + +public class GroupedPercentileAccumulator implements GroupedAccumulator { + private static final long INSTANCE_SIZE = + RamUsageEstimator.shallowSizeOfInstance(GroupedPercentileAccumulator.class); + private final TSDataType seriesDataType; + private double percentage; + private final MemoryReservationManager memoryReservationManager; + private long previousArraySize; + private final PercentileBigArray array = new PercentileBigArray(); + + public GroupedPercentileAccumulator( + TSDataType seriesDataType, MemoryReservationManager memoryReservationManager) { + this.seriesDataType = seriesDataType; + this.memoryReservationManager = memoryReservationManager; + updateMemoryReservation(); + } + + @Override + public long getEstimatedSize() { + return INSTANCE_SIZE + array.sizeOf(); + } + + @Override + public void setGroupCount(long groupCount) { + array.ensureCapacity(groupCount); + } + + @Override + public void addInput(int[] groupIds, Column[] arguments, AggregationMask mask) { + if (arguments.length != 2) { + throw new SemanticException( + String.format("PERCENTILE requires 2 arguments, but got %d", arguments.length)); + } + percentage = arguments[1].getDouble(0); + + switch (seriesDataType) { + case INT32: + addIntInput(groupIds, arguments, mask); + break; + case INT64: + case TIMESTAMP: + addLongInput(groupIds, arguments, mask); + break; + case FLOAT: + addFloatInput(groupIds, arguments, mask); + break; + case DOUBLE: + addDoubleInput(groupIds, arguments, mask); + break; + default: + throw new UnSupportedDataTypeException( + String.format("Unsupported data type in PERCENTILE Aggregation: %s", seriesDataType)); + } + updateMemoryReservation(); + } + + @Override + public void addIntermediate(int[] groupIds, Column argument) { + for (int i = 0; i < groupIds.length; i++) { + int groupId = groupIds[i]; + if (!argument.isNull(i)) { + byte[] data = argument.getBinary(i).getValues(); + ByteBuffer buffer = ByteBuffer.wrap(data); + this.percentage = ReadWriteIOUtils.readDouble(buffer); + Percentile other = Percentile.deserialize(buffer); + array.get(groupId).merge(other); + } + } + updateMemoryReservation(); + } + + @Override + public void evaluateIntermediate(int groupId, ColumnBuilder columnBuilder) { + Percentile percentile = array.get(groupId); + int percentileDataLength = percentile.getSerializedSize(); + ByteBuffer buffer = ByteBuffer.allocate(8 + percentileDataLength); + ReadWriteIOUtils.write(percentage, buffer); + percentile.serialize(buffer); + columnBuilder.writeBinary(new Binary(buffer.array())); + } + + @Override + public void evaluateFinal(int groupId, ColumnBuilder columnBuilder) { + Percentile percentile = array.get(groupId); + double result = percentile.getPercentile(percentage); + if (Double.isNaN(result)) { + columnBuilder.appendNull(); + return; + } + switch (seriesDataType) { + case INT32: + columnBuilder.writeInt((int) result); + break; + case INT64: + case TIMESTAMP: + columnBuilder.writeLong((long) result); + break; + case FLOAT: + columnBuilder.writeFloat((float) result); + break; + case DOUBLE: + columnBuilder.writeDouble(result); + break; + default: + throw new UnSupportedDataTypeException( + String.format("Unsupported data type in PERCENTILE Aggregation: %s", seriesDataType)); + } + } + + @Override + public void prepareFinal() {} + + @Override + public void reset() { + array.reset(); + updateMemoryReservation(); + } + + private void updateMemoryReservation() { + long currentSize = array.sizeOf(); + long delta = currentSize - previousArraySize; + if (delta > 0) { + memoryReservationManager.reserveMemoryCumulatively(delta); + } else if (delta < 0) { + memoryReservationManager.releaseMemoryCumulatively(-delta); + } + previousArraySize = currentSize; + } + + public void addIntInput(int[] groupIds, Column[] arguments, AggregationMask mask) { + Column valueColumn = arguments[0]; + + int positionCount = mask.getPositionCount(); Review Comment: 🔴 **[正确性 / 阻塞]** 这里用了 `mask.getPositionCount()`，但 `else`（非 selectAll）分支的循环边界应当是 `mask.getSelectedPositionCount()`。 当聚合带过滤掩码时（如 `percentile(x, 0.5) FILTER (WHERE ...)`，或优化器引入的 mask channel——见 `GroupedAggregator` 里 `applyMaskBlock` 的调用），`mask.getSelectedPositions()` 数组只有前 `selectedPositionCount` 个元素有效，其余是上一批的残留或 0。按 `positionCount`（> `selectedPositionCount`）遍历会读到这些无效下标：轻则把 position 0 等错误位置重复计入，导致**分组结果静默错误**；若底层数组长度不足则抛 `ArrayIndexOutOfBoundsException`。 对照本 PR 的非分组版 `PercentileAccumulator.addIntInput` 以及现有的 `GroupedModeAccumulator`/`GroupedAvgAccumulator`，全部用的是 `getSelectedPositionCount()`。selectAll 分支因 `selectedPositionCount == positionCount` 不受影响，所以直接改成 `getSelectedPositionCount()` 即可让两个分支都正确。 **相同问题存在于 `addLongInput`(L192)、`addFloatInput`(L220)、`addDoubleInput`(L248)。** 现有 IT 未覆盖带 FILTER 的 percentile，因此没被测出来，建议补一条用例。 另：这四个 `add*Input` 方法建议改为 `private`（同类 accumulator 均为 private）。 ########## iotdb-core/calc-commons/src/main/java/org/apache/iotdb/calc/execution/operator/source/relational/Percentile.java: ########## @@ -0,0 +1,161 @@ +/* Review Comment: 🟡 **[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`。直接复制同目录现有文件的头即可。 ########## iotdb-core/calc-commons/src/main/java/org/apache/iotdb/calc/execution/operator/source/relational/Percentile.java: ########## @@ -0,0 +1,161 @@ +/* + * Licensed 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() { + 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) { Review Comment: 🟡 **[健壮性 / 隐患]** `deserialize` 出来的对象 `sorted` 仍是构造函数里的 `true`，但 `serialize()` 写出的数据并不保证有序（序列化时通常 `sorted == false`）。也就是说反序列化得到的 `Percentile` 自称“已排序”，实则可能无序。 当前尚未真正触发 bug，因为反序列化结果只通过 `merge()` 被消费（merge 直接读 `values[]` 并把目标置为 `sorted = false`）。但这是个定时炸弹：一旦将来出现直接对反序列化对象调用 `getPercentile()` 的路径（例如单 partial 跳过 merge 的优化），`ensureSorted()` 会因 `sorted == true` 跳过排序而**静默返回错误结果**。 建议在此显式设置 `percentile.sorted = false;`（或让 `serialize()` 前先 `ensureSorted()`，使序列化数据始终有序）。 -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
