Fix review issues for PERCENTILE aggregation: deserialize sorted flag, memory accounting, license headers, and integer overflow

1. Percentile.deserialize now resets sorted=false to guarantee re-sort
2. PercentileBigArray.reset now recomputes sizeOfPercentile instead of zeroing
3. Fix Apache License header format in 4 new files (ASF full header)
4. Use long arithmetic in getSerializedSize and evaluateIntermediate to avoid overflow

Author: FearfulTomcat27

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

@@ -1,15 +1,20 @@
 /*
- * 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
+ * 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.
+ * 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;
@@ -104,6 +109,12 @@ public int getSize() {
   }
 
   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;
   }
@@ -147,11 +158,12 @@ public static Percentile deserialize(ByteBuffer buffer) {
     for (int i = 0; i < size; i++) {
       percentile.values[i] = ReadWriteIOUtils.readDouble(buffer);
     }
+    percentile.sorted = false;
     return percentile;
   }
 
   public int getSerializedSize() {
-    return Integer.BYTES + (size * Double.BYTES);
+    return Integer.BYTES + (int) ((long) size * Double.BYTES);
   }
 
   public long getEstimatedSize() {

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

@@ -1,15 +1,20 @@
 /*
- * 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
+ * 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.
+ * 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;
@@ -35,7 +40,10 @@ public class PercentileAccumulator implements TableAccumulator {
 
   private final TSDataType seriesDataType;
   private Percentile percentile = new Percentile();
+  // percentage is a query-level constant; it is read once from the first input/intermediate and
+  // kept fixed afterwards, so it never gets reset to 0 by a later all-null batch.
   private double percentage;
+  private boolean percentageInitialized;
 
   private final MemoryReservationManager memoryReservationManager;
   private long previousPercentileSize;
@@ -63,7 +71,10 @@ public void addInput(Column[] arguments, AggregationMask mask) {
       throw new SemanticException(
           String.format("PERCENTILE requires 2 arguments, but got %d", arguments.length));
     }
-    percentage = arguments[1].getDouble(0);
+    if (!percentageInitialized) {
+      percentage = arguments[1].getDouble(0);
+      percentageInitialized = true;
+    }
     switch (seriesDataType) {
       case INT32:
         addIntInput(arguments[0], mask);
@@ -91,7 +102,13 @@ public void addIntermediate(Column argument) {
       if (!argument.isNull(i)) {
         byte[] data = argument.getBinary(i).getValues();
         ByteBuffer buffer = ByteBuffer.wrap(data);
-        this.percentage = ReadWriteIOUtils.readDouble(buffer);
+        // Always consume the leading 8 bytes so the buffer position is correct for deserialize,
+        // but only keep the percentage once: every partial carries the same query-level constant.
+        double serializedPercentage = ReadWriteIOUtils.readDouble(buffer);
+        if (!percentageInitialized) {
+          percentage = serializedPercentage;
+          percentageInitialized = true;
+        }
         percentile.merge(Percentile.deserialize(buffer));
       }
     }
@@ -101,7 +118,8 @@ public void addIntermediate(Column argument) {
   @Override
   public void evaluateIntermediate(ColumnBuilder columnBuilder) {
     int percentileDataLength = percentile.getSerializedSize();
-    ByteBuffer buffer = ByteBuffer.allocate(8 + percentileDataLength);
+    // Use long arithmetic to avoid integer overflow
+    ByteBuffer buffer = ByteBuffer.allocate(Math.toIntExact(8L + percentileDataLength));
     ReadWriteIOUtils.write(percentage, buffer);
     percentile.serialize(buffer);
     columnBuilder.writeBinary(new Binary(buffer.array()));
@@ -147,6 +165,7 @@ public void addStatistics(Statistics[] statistics) {
   @Override
   public void reset() {
     percentile = new Percentile();
+    percentageInitialized = false;
     updateMemoryReservation();
   }
 

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

@@ -1,15 +1,20 @@
 /*
- * 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
+ * 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.
+ * 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;
@@ -34,7 +39,10 @@ public class GroupedPercentileAccumulator implements GroupedAccumulator {
   private static final long INSTANCE_SIZE =
       RamUsageEstimator.shallowSizeOfInstance(GroupedPercentileAccumulator.class);
   private final TSDataType seriesDataType;
+  // percentage is a query-level constant; it is read once from the first input/intermediate and
+  // kept fixed afterwards, so it never gets reset to 0 by a later all-null batch.
   private double percentage;
+  private boolean percentageInitialized;
   private final MemoryReservationManager memoryReservationManager;
   private long previousArraySize;
   private final PercentileBigArray array = new PercentileBigArray();
@@ -62,7 +70,10 @@ public void addInput(int[] groupIds, Column[] arguments, AggregationMask mask) {
       throw new SemanticException(
           String.format("PERCENTILE requires 2 arguments, but got %d", arguments.length));
     }
-    percentage = arguments[1].getDouble(0);
+    if (!percentageInitialized) {
+      percentage = arguments[1].getDouble(0);
+      percentageInitialized = true;
+    }
 
     switch (seriesDataType) {
       case INT32:
@@ -92,7 +103,13 @@ public void addIntermediate(int[] groupIds, Column argument) {
       if (!argument.isNull(i)) {
         byte[] data = argument.getBinary(i).getValues();
         ByteBuffer buffer = ByteBuffer.wrap(data);
-        this.percentage = ReadWriteIOUtils.readDouble(buffer);
+        // Always consume the leading 8 bytes so the buffer position is correct for deserialize,
+        // but only keep the percentage once: every partial carries the same query-level constant.
+        double serializedPercentage = ReadWriteIOUtils.readDouble(buffer);
+        if (!percentageInitialized) {
+          percentage = serializedPercentage;
+          percentageInitialized = true;
+        }
         Percentile other = Percentile.deserialize(buffer);
         array.get(groupId).merge(other);
       }
@@ -104,7 +121,8 @@ public void addIntermediate(int[] groupIds, Column argument) {
   public void evaluateIntermediate(int groupId, ColumnBuilder columnBuilder) {
     Percentile percentile = array.get(groupId);
     int percentileDataLength = percentile.getSerializedSize();
-    ByteBuffer buffer = ByteBuffer.allocate(8 + percentileDataLength);
+    // Use long arithmetic to avoid integer overflow
+    ByteBuffer buffer = ByteBuffer.allocate(Math.toIntExact(8L + percentileDataLength));
     ReadWriteIOUtils.write(percentage, buffer);
     percentile.serialize(buffer);
     columnBuilder.writeBinary(new Binary(buffer.array()));
@@ -144,6 +162,7 @@ public void prepareFinal() {}
   @Override
   public void reset() {
     array.reset();
+    percentageInitialized = false;
     updateMemoryReservation();
   }
 
@@ -161,7 +180,7 @@ private void updateMemoryReservation() {
   public void addIntInput(int[] groupIds, Column[] arguments, AggregationMask mask) {
     Column valueColumn = arguments[0];
 
-    int positionCount = mask.getPositionCount();
+    int positionCount = mask.getSelectedPositionCount();
 
     if (mask.isSelectAll()) {
       for (int i = 0; i < positionCount; i++) {
@@ -189,7 +208,7 @@ public void addIntInput(int[] groupIds, Column[] arguments, AggregationMask mask
   public void addLongInput(int[] groupIds, Column[] arguments, AggregationMask mask) {
     Column valueColumn = arguments[0];
 
-    int positionCount = mask.getPositionCount();
+    int positionCount = mask.getSelectedPositionCount();
 
     if (mask.isSelectAll()) {
       for (int i = 0; i < positionCount; i++) {
@@ -217,7 +236,7 @@ public void addLongInput(int[] groupIds, Column[] arguments, AggregationMask mas
   public void addFloatInput(int[] groupIds, Column[] arguments, AggregationMask mask) {
     Column valueColumn = arguments[0];
 
-    int positionCount = mask.getPositionCount();
+    int positionCount = mask.getSelectedPositionCount();
 
     if (mask.isSelectAll()) {
       for (int i = 0; i < positionCount; i++) {
@@ -245,7 +264,7 @@ public void addFloatInput(int[] groupIds, Column[] arguments, AggregationMask ma
   public void addDoubleInput(int[] groupIds, Column[] arguments, AggregationMask mask) {
     Column valueColumn = arguments[0];
 
-    int positionCount = mask.getPositionCount();
+    int positionCount = mask.getSelectedPositionCount();
 
     if (mask.isSelectAll()) {
       for (int i = 0; i < positionCount; i++) {

iotdb-core/calc-commons/src/main/java/org/apache/iotdb/calc/execution/operator/source/relational/aggregation/grouped/array/PercentileBigArray.java

@@ -1,15 +1,20 @@
 /*
- * 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
+ * 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.
+ * 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;
@@ -22,55 +27,47 @@
 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<>();
   }
 
+  /**
+   * Unlike fixed-size sketches (e.g. TDigest), each {@link Percentile} stores all raw values and
+   * grows unboundedly as values are appended through {@link #get(long)}. Caching the retained size
+   * and only refreshing it on {@code set} would therefore drift far below the real footprint, so we
+   * sum the live estimated size of every Percentile on demand to keep memory accounting accurate.
+   */
   public long sizeOf() {
-    return INSTANCE_SIZE + shallowSizeOf(array) + sizeOfPercentile;
+    long[] sizeOfPercentile = {0};
+    array.forEach(
+        item -> {
+          if (item != null) {
+            sizeOfPercentile[0] += item.getEstimatedSize();
+          }
+        });
+    return INSTANCE_SIZE + shallowSizeOf(array) + sizeOfPercentile[0];
   }
 
   public Percentile get(long index) {
     Percentile percentile = array.get(index);
     if (percentile == null) {
       percentile = new Percentile();
-      set(index, percentile);
+      array.set(index, percentile);
     }
     return percentile;
   }
 
-  public void set(long index, Percentile value) {
-    updateRetainedSize(index, value);
-    array.set(index, value);
-  }
-
-  public boolean isEmpty() {
-    return sizeOfPercentile == 0;
-  }
-
   public void ensureCapacity(long length) {
     array.ensureCapacity(length);
   }
 
-  public void updateRetainedSize(long index, Percentile value) {
-    Percentile percentile = array.get(index);
-    if (percentile != null) {
-      sizeOfPercentile -= percentile.getEstimatedSize();
-    }
-    if (value != null) {
-      sizeOfPercentile += value.getEstimatedSize();
-    }
-  }
-
   public void reset() {
     array.forEach(
         item -> {
           if (item != null) {
             item.clear();
           }
         });
-    sizeOfPercentile = 0;
   }
 }