[CALCITE-6767] PERCENTILE_CONT/PERCENTILE_DISC function are not supported

core/src/main/java/org/apache/calcite/adapter/enumerable/EnumerableAggregateBase.java

@@ -30,6 +30,7 @@
 import org.apache.calcite.plan.RelOptCluster;
 import org.apache.calcite.plan.RelTraitSet;
 import org.apache.calcite.rel.RelCollations;
+import org.apache.calcite.rel.RelFieldCollation;
 import org.apache.calcite.rel.RelNode;
 import org.apache.calcite.rel.core.Aggregate;
 import org.apache.calcite.rel.core.AggregateCall;
@@ -233,7 +234,7 @@
     }
   }
 
   protected void createAccumulatorAdders(
       final ParameterExpression inParameter,
       final List<AggImpState> aggs,
       final PhysType accPhysType,
@@ -264,6 +265,20 @@
               for (int index : agg.call.getArgList()) {
                 args.add(RexInputRef.of(index, inputTypes));
               }
+              // Percentile functions such as PERCENTILE_CONT and
+              // PERCENTILE_DISC take the fraction as their only argument, but
+              // aggregate over the WITHIN GROUP (ORDER BY ...) column. Expose
+              // that collation column as an extra argument so that the
+              // accumulator can collect its values (already sorted by
+              // SourceSorter).
+              if (agg.call.getAggregation().isPercentile()) {
+                for (RelFieldCollation fieldCollation
+                    : agg.call.collation.getFieldCollations()) {
+                  args.add(
+                      RexInputRef.of(fieldCollation.getFieldIndex(),
+                          inputTypes));
+                }
+              }
               return args;
             }
 

core/src/main/java/org/apache/calcite/adapter/enumerable/RexImpTable.java

@@ -499,6 +499,8 @@
 import static org.apache.calcite.sql.fun.SqlStdOperatorTable.OCTET_LENGTH;
 import static org.apache.calcite.sql.fun.SqlStdOperatorTable.OR;
 import static org.apache.calcite.sql.fun.SqlStdOperatorTable.OVERLAY;
+import static org.apache.calcite.sql.fun.SqlStdOperatorTable.PERCENTILE_CONT;
+import static org.apache.calcite.sql.fun.SqlStdOperatorTable.PERCENTILE_DISC;
 import static org.apache.calcite.sql.fun.SqlStdOperatorTable.PI;
 import static org.apache.calcite.sql.fun.SqlStdOperatorTable.PLUS;
 import static org.apache.calcite.sql.fun.SqlStdOperatorTable.POSITION;
@@ -1321,6 +1323,8 @@
       defineAgg(SINGLE_VALUE, SingleValueImplementor.class);
       defineAgg(COLLECT, CollectImplementor.class);
       defineAgg(ARRAY_AGG, CollectImplementor.class);
+      defineAgg(PERCENTILE_CONT, PercentileImplementor.class);
+      defineAgg(PERCENTILE_DISC, PercentileImplementor.class);
       defineAgg(LISTAGG, ListaggImplementor.class);
       defineAgg(FUSION, FusionImplementor.class);
       defineAgg(MODE, ModeImplementor.class);
@@ -1467,7 +1471,7 @@
     };
   }
 
   public @Nullable RexCallImplementor get(final SqlOperator operator) {
     if (operator instanceof SqlUserDefinedFunction) {
       org.apache.calcite.schema.Function udf =
           ((SqlUserDefinedFunction) operator).getFunction();
@@ -1966,6 +1970,60 @@
     }
   }
 
+  /** Implementor for the {@code PERCENTILE_CONT} and {@code PERCENTILE_DISC}
+   * aggregate functions.
+   *
+   * <p>The fraction is the sole argument of the aggregate call, while the
+   * values whose percentile is computed come from the
+   * {@code WITHIN GROUP (ORDER BY ...)} column, which
+   * {@link EnumerableAggregateBase#createAccumulatorAdders} exposes as an extra
+   * argument. The input rows are sorted by {@code SourceSorter} before being
+   * accumulated, so the collected values are already in order. */
+  static class PercentileImplementor extends StrictAggImplementor {
+    @Override public List<Type> getNotNullState(AggContext info) {
+      final List<Type> types = new ArrayList<>();
+      types.add(List.class);
+      types.add(double.class);
+      return types;
+    }
+
+    @Override protected void implementNotNullReset(AggContext info,
+        AggResetContext reset) {
+      reset.currentBlock().add(
+          Expressions.statement(
+              Expressions.assign(reset.accumulator().get(0),
+                  Expressions.new_(ArrayList.class))));
+      reset.currentBlock().add(
+          Expressions.statement(
+              Expressions.assign(reset.accumulator().get(1),
+                  Expressions.constant(0d))));
+    }
+
+    @Override protected void implementNotNullAdd(AggContext info,
+        AggAddContext add) {
+      add.currentBlock().add(
+          Expressions.statement(
+              Expressions.assign(add.accumulator().get(1),
+                  EnumUtils.convert(add.arguments().get(0), double.class))));
+
+      add.currentBlock().add(
+          Expressions.statement(
+              Expressions.call(add.accumulator().get(0),
+                  BuiltInMethod.COLLECTION_ADD.method,
+                  Expressions.box(add.arguments().get(1)))));
+    }
+
+    @Override protected Expression implementNotNullResult(AggContext info,
+        AggResultContext result) {
+      final BuiltInMethod method =
+          info.aggregation().kind == SqlKind.PERCENTILE_DISC
+              ? BuiltInMethod.PERCENTILE_DISC
+              : BuiltInMethod.PERCENTILE_CONT;
+      return Expressions.call(method.method, result.accumulator().get(0),
+          result.accumulator().get(1));
+    }
+  }
+
   /** Implementor for the {@code LISTAGG} aggregate function. */
   static class ListaggImplementor extends StrictAggImplementor {
     @Override protected void implementNotNullReset(AggContext info,

core/src/main/java/org/apache/calcite/runtime/SqlFunctions.java

@@ -133,6 +133,7 @@
 import java.util.List;
 import java.util.Locale;
 import java.util.Map;
+import java.util.NoSuchElementException;
 import java.util.Objects;
 import java.util.Set;
 import java.util.TimeZone;
@@ -4053,6 +4054,55 @@ public static BigDecimal mod(BigDecimal b0, BigDecimal b1) {
     return bigDecimals[1];
   }
 
+  // PERCENTILE_CONT / PERCENTILE_DISC
+
+  /** Support the PERCENTILE_CONT aggregate function.
+   *
+   * <p>The {@code values} list must already be sorted according to the
+   * {@code WITHIN GROUP (ORDER BY ...)} clause. The fraction must be in the
+   * range 0 to 1 inclusive. The result is a linear interpolation between the
+   * two values that surround the desired position. */
+  public static BigDecimal percentileCont(List<? extends Number> values,
+      double fraction) {
+    final int n = values.size();
+    if (n == 0) {
+      throw new NoSuchElementException(
+          "PERCENTILE_CONT is not defined on an empty group");
+    }
+    final double rank = fraction * (n - 1);
+    final int lo = (int) Math.floor(rank);
+    final int hi = (int) Math.ceil(rank);
+    final BigDecimal loValue = toBigDecimal(values.get(lo));
+    if (lo == hi) {
+      return loValue;
+    }
+    final BigDecimal hiValue = toBigDecimal(values.get(hi));
+    final BigDecimal frac = BigDecimal.valueOf(rank - lo);
+    return loValue.add(hiValue.subtract(loValue).multiply(frac));
+  }
+
+  /** Support the PERCENTILE_DISC aggregate function.
+   *
+   * <p>The {@code values} list must already be sorted according to the
+   * {@code WITHIN GROUP (ORDER BY ...)} clause. The fraction must be in the
+   * range 0 to 1 inclusive. The result is an actual value from the group: the
+   * first whose cumulative distribution is greater than or equal to the
+   * fraction. */
+  public static Object percentileDisc(List<?> values, double fraction) {
+    final int n = values.size();
+    if (n == 0) {
+      throw new NoSuchElementException(
+          "PERCENTILE_DISC is not defined on an empty group");
+    }
+    int index = (int) Math.ceil(fraction * n) - 1;
+    if (index < 0) {
+      index = 0;
+    } else if (index >= n) {
+      index = n - 1;
+    }
+    return requireNonNull(values.get(index));
+  }
+
   // FLOOR
 
   public static double floor(double b0) {

core/src/main/java/org/apache/calcite/util/BuiltInMethod.java

@@ -408,6 +408,8 @@ public enum BuiltInMethod {
   COLLECTION_ADD(Collection.class, "add", Object.class),
   COLLECTION_ADDALL(Collection.class, "addAll", Collection.class),
   COLLECTION_RETAIN_ALL(Collection.class, "retainAll", Collection.class),
+  PERCENTILE_CONT(SqlFunctions.class, "percentileCont", List.class, double.class),
+  PERCENTILE_DISC(SqlFunctions.class, "percentileDisc", List.class, double.class),
   LIST_CONTAINS(List.class, "contains", Object.class),
   LIST_GET(List.class, "get", int.class),
   LIST_TO_ARRAY(List.class, "toArray"),

core/src/test/resources/sql/agg.iq

@@ -3844,6 +3844,119 @@ select distinct sum(deptno + '1') as deptsum from dept order by 1;
 
 !ok
 
+# [CALCITE-6767] PERCENTILE_CONT/PERCENTILE_DISC syntax not supported.
+# These sql programs were validated in PostgreSQL
+select
+  percentile_cont(0.5) within group (order by empno) as c,
+  percentile_disc(0.5) within group (order by empno) as d
+from emp;
++------+------+
+| C    | D    |
++------+------+
+| 7785 | 7782 |
++------+------+
+(1 row)
+
+!ok
+
+# PERCENTILE_CONT / PERCENTILE_DISC with GROUP BY.
+select deptno,
+  percentile_cont(0.5) within group (order by empno) as c,
+  percentile_disc(0.5) within group (order by empno) as d
+from emp
+group by deptno
+order by deptno;
++--------+------+------+
+| DEPTNO | C    | D    |
++--------+------+------+
+|     10 | 7839 | 7839 |
+|     20 | 7788 | 7788 |
+|     30 | 7676 | 7654 |
++--------+------+------+
+(3 rows)
+
+!ok
+
+# PERCENTILE_CONT / PERCENTILE_DISC honour a descending ORDER BY.
+select
+  percentile_disc(0.25) within group (order by empno desc) as d
+from emp;
++------+
+| D    |
++------+
+| 7876 |
++------+
+(1 row)
+
+!ok
+
+# PERCENTILE_CONT / PERCENTILE_DISC on DOUBLE values.
+select
+  percentile_cont(0.5) within group (order by v) as c,
+  percentile_disc(0.5) within group (order by v) as d
+from (values (cast(1.5 as double)),
+             (cast(2.5 as double)),
+             (cast(4.5 as double)),
+             (cast(8.5 as double))) as t(v);
++-----+-----+
+| C   | D   |
++-----+-----+
+| 3.5 | 2.5 |
++-----+-----+
+(1 row)
+
+!ok
+
+# PERCENTILE_CONT / PERCENTILE_DISC on large DECIMAL values that are very
+# close together. These 19-digit values exceed the ~15-16 significant digits a
+# double can hold, so the linear interpolation must be done in BigDecimal;
+# converting to double would round 9999999999999999990 to 1.0E19 and lose the
+# trailing digits.
+select
+  percentile_cont(0.5) within group (order by v) as c,
+  percentile_disc(0.5) within group (order by v) as d
+from (values (cast('9999999999999999990' as decimal(19, 0))),
+             (cast('9999999999999999992' as decimal(19, 0)))) as t(v);
++-----------------------+---------------------+
+| C                     | D                   |
++-----------------------+---------------------+
+| 9999999999999999991.0 | 9999999999999999990 |
++-----------------------+---------------------+
+(1 row)
+
+!ok
+
+# A three-row large DECIMAL group whose 0.75 percentile interpolates between
+# two adjacent, nearly-equal values.
+select
+  percentile_cont(0.75) within group (order by v) as c
+from (values (cast('9999999999999999990' as decimal(19, 0))),
+             (cast('9999999999999999994' as decimal(19, 0))),
+             (cast('9999999999999999998' as decimal(19, 0)))) as t(v);
++-----------------------+
+| C                     |
++-----------------------+
+| 9999999999999999996.0 |
++-----------------------+
+(1 row)
+
+!ok
+
+# PERCENTILE_CONT / PERCENTILE_DISC on UNSIGNED values.
+select
+  percentile_cont(0.5) within group (order by v) as c,
+  percentile_disc(0.5) within group (order by v) as d
+from (values (cast(10 as int unsigned)),
+             (cast(20 as int unsigned))) as t(v);
++----+----+
+| C  | D  |
++----+----+
+| 15 | 10 |
++----+----+
+(1 row)
+
+!ok
+
 # [CALCITE-6839] The SUM function sometimes throws overflow exceptions due
 # to incorrect return types
 !use scott-spark