Add PERCENTILE aggregation function and related validation

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

@@ -4345,6 +4345,42 @@ public void approxPercentileTest() {
         DATABASE_NAME);
   }
 
+  @Test
+  public void percentileTest() {
+    tableResultSetEqualTest(
+        "select percentile(time, 0.5),percentile(s1,0.5),percentile(s2,0.5),percentile(s3,0.5),percentile(s4,0.5),percentile(s9,0.5) from table1",
+        buildHeaders(6),
+        new String[] {"2024-09-24T06:15:40.000Z,40,43000,37.5,43.0,2024-09-24T06:15:40.000Z,"},
+        DATABASE_NAME);
+
+    tableResultSetEqualTest(
+        "select time,province,percentile(time, 0.5),percentile(s1,0.5),percentile(s2,0.5) from table1 group by 1,2 order by 2,1",
+        new String[] {"time", "province", "_col2", "_col3", "_col4"},
+        new String[] {
+          "2024-09-24T06:15:30.000Z,beijing,2024-09-24T06:15:30.000Z,30,null,",
+          "2024-09-24T06:15:31.000Z,beijing,2024-09-24T06:15:31.000Z,null,31000,",
+          "2024-09-24T06:15:35.000Z,beijing,2024-09-24T06:15:35.000Z,null,35000,",
+          "2024-09-24T06:15:36.000Z,beijing,2024-09-24T06:15:36.000Z,36,null,",
+          "2024-09-24T06:15:40.000Z,beijing,2024-09-24T06:15:40.000Z,40,40000,",
+          "2024-09-24T06:15:41.000Z,beijing,2024-09-24T06:15:41.000Z,41,null,",
+          "2024-09-24T06:15:46.000Z,beijing,2024-09-24T06:15:46.000Z,null,46000,",
+          "2024-09-24T06:15:50.000Z,beijing,2024-09-24T06:15:50.000Z,null,50000,",
+          "2024-09-24T06:15:51.000Z,beijing,2024-09-24T06:15:51.000Z,null,null,",
+          "2024-09-24T06:15:55.000Z,beijing,2024-09-24T06:15:55.000Z,55,null,",
+          "2024-09-24T06:15:30.000Z,shanghai,2024-09-24T06:15:30.000Z,30,null,",
+          "2024-09-24T06:15:31.000Z,shanghai,2024-09-24T06:15:31.000Z,null,31000,",
+          "2024-09-24T06:15:35.000Z,shanghai,2024-09-24T06:15:35.000Z,null,35000,",
+          "2024-09-24T06:15:36.000Z,shanghai,2024-09-24T06:15:36.000Z,36,null,",
+          "2024-09-24T06:15:40.000Z,shanghai,2024-09-24T06:15:40.000Z,40,40000,",
+          "2024-09-24T06:15:41.000Z,shanghai,2024-09-24T06:15:41.000Z,41,null,",
+          "2024-09-24T06:15:46.000Z,shanghai,2024-09-24T06:15:46.000Z,null,46000,",
+          "2024-09-24T06:15:50.000Z,shanghai,2024-09-24T06:15:50.000Z,null,50000,",
+          "2024-09-24T06:15:51.000Z,shanghai,2024-09-24T06:15:51.000Z,null,null,",
+          "2024-09-24T06:15:55.000Z,shanghai,2024-09-24T06:15:55.000Z,55,null,",
+        },
+        DATABASE_NAME);
+  }
+
   @Test
   public void exceptionTest() {
     tableAssertTestFail(
@@ -4423,6 +4459,22 @@ public void exceptionTest() {
         "select approx_percentile(s5,0.5) from table1",
         "701: Aggregation functions [approx_percentile] should have value column as numeric type [INT32, INT64, FLOAT, DOUBLE, TIMESTAMP]",
         DATABASE_NAME);
+    tableAssertTestFail(
+        "select percentile() from table1",
+        "701: Aggregation functions [percentile] should only have two arguments",
+        DATABASE_NAME);
+    tableAssertTestFail(
+        "select percentile(s1,1.1) from table1",
+        "701: percentage should be in [0,1], got 1.1",
+        DATABASE_NAME);
+    tableAssertTestFail(
+        "select percentile(s1,'test') from table1",
+        "701: The second argument of 'percentile' function percentage must be a double literal",
+        DATABASE_NAME);
+    tableAssertTestFail(
+        "select percentile(s5,0.5) from table1",
+        "701: Aggregation functions [percentile] should have value column as numeric type [INT32, INT64, FLOAT, DOUBLE, TIMESTAMP]",
+        DATABASE_NAME);
   }
 
   // ==================================================================

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

@@ -0,0 +1,164 @@
+/*
+ * 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.db.queryengine.execution.operator.source.relational;
+
+import org.apache.iotdb.db.exception.sql.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);
+    ReadWriteIOUtils.write(capacity, buffer);
+    ReadWriteIOUtils.write(sorted, buffer);
+    for (int i = 0; i < size; i++) {
+      ReadWriteIOUtils.write(values[i], buffer);
+    }
+  }
+
+  public static Percentile deserialize(ByteBuffer buffer) {
+    Percentile percentile = new Percentile();
+    percentile.size = ReadWriteIOUtils.readInt(buffer);
+    percentile.capacity = ReadWriteIOUtils.readInt(buffer);
+    percentile.sorted = ReadWriteIOUtils.readBool(buffer);
+
+    if (percentile.capacity != percentile.values.length) {
+      percentile.values = new double[percentile.capacity];
+    }
+
+    for (int i = 0; i < percentile.size; i++) {
+      percentile.values[i] = ReadWriteIOUtils.readDouble(buffer);
+    }
+
+    return percentile;
+  }
+
+  public int getSerializedSize() {
+    return Integer.BYTES + Integer.BYTES + 1 + (size * Double.BYTES);
+  }
+
+  public long getEstimatedSize() {
+    long shallowSize = RamUsageEstimator.shallowSizeOfInstance(Percentile.class);
+    return shallowSize + getSerializedSize();
+  }
+}

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

@@ -45,6 +45,7 @@
 import org.apache.iotdb.db.queryengine.execution.operator.source.relational.aggregation.grouped.GroupedMinAccumulator;
 import org.apache.iotdb.db.queryengine.execution.operator.source.relational.aggregation.grouped.GroupedMinByAccumulator;
 import org.apache.iotdb.db.queryengine.execution.operator.source.relational.aggregation.grouped.GroupedModeAccumulator;
+import org.apache.iotdb.db.queryengine.execution.operator.source.relational.aggregation.grouped.GroupedPercentileAccumulator;
 import org.apache.iotdb.db.queryengine.execution.operator.source.relational.aggregation.grouped.GroupedSumAccumulator;
 import org.apache.iotdb.db.queryengine.execution.operator.source.relational.aggregation.grouped.GroupedUserDefinedAggregateAccumulator;
 import org.apache.iotdb.db.queryengine.execution.operator.source.relational.aggregation.grouped.GroupedVarianceAccumulator;
@@ -283,6 +284,8 @@ private static GroupedAccumulator createBuiltinGroupedAccumulator(
         } else {
           return new GroupedApproxPercentileWithWeightAccumulator(inputDataTypes.get(0));
         }
+      case PERCENTILE:
+        return new GroupedPercentileAccumulator(inputDataTypes.get(0));
       default:
         throw new IllegalArgumentException("Invalid Aggregation function: " + aggregationType);
     }
@@ -367,6 +370,8 @@ public static TableAccumulator createBuiltinAccumulator(
         } else {
           return new ApproxPercentileWithWeightAccumulator(inputDataTypes.get(0));
         }
+      case PERCENTILE:
+        return new PercentileAccumulator(inputDataTypes.get(0));
       default:
         throw new IllegalArgumentException("Invalid Aggregation function: " + aggregationType);
     }