Skip to content

Commit a9a891a

Browse files
LeomrlinLeomrlin
authored
Support percentile (#534)
1 parent fd806ec commit a9a891a

8 files changed

Lines changed: 382 additions & 0 deletions

File tree

‎geaflow/geaflow-dsl/geaflow-dsl-plan/src/main/java/com/antgroup/geaflow/dsl/schema/function/BuildInSqlFunctionTable.java‎

Lines changed: 7 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -34,6 +34,9 @@
3434
import com.antgroup.geaflow.dsl.udf.graph.SingleSourceShortestPath;
3535
import com.antgroup.geaflow.dsl.udf.graph.TriangleCount;
3636
import com.antgroup.geaflow.dsl.udf.graph.WeakConnectedComponents;
37+
import com.antgroup.geaflow.dsl.udf.table.agg.PercentileDouble;
38+
import com.antgroup.geaflow.dsl.udf.table.agg.PercentileInteger;
39+
import com.antgroup.geaflow.dsl.udf.table.agg.PercentileLong;
3740
import com.antgroup.geaflow.dsl.udf.table.date.AddMonths;
3841
import com.antgroup.geaflow.dsl.udf.table.date.DateAdd;
3942
import com.antgroup.geaflow.dsl.udf.table.date.DateDiff;
@@ -182,6 +185,10 @@ public class BuildInSqlFunctionTable extends ListSqlOperatorTable {
182185
.add(GeaFlowFunction.of(EdgeTargetId.class))
183186
.add(GeaFlowFunction.of(EdgeTimestamp.class))
184187
.add(GeaFlowFunction.of(IsDecimal.class))
188+
// UDAF
189+
.add(GeaFlowFunction.of(PercentileLong.class))
190+
.add(GeaFlowFunction.of(PercentileInteger.class))
191+
.add(GeaFlowFunction.of(PercentileDouble.class))
185192
// UDGA
186193
.add(GeaFlowFunction.of(SingleSourceShortestPath.class))
187194
.add(GeaFlowFunction.of(AllSourceShortestPath.class))

‎geaflow/geaflow-dsl/geaflow-dsl-plan/src/main/java/com/antgroup/geaflow/dsl/udf/table/agg/PercentileDouble.java‎

Lines changed: 136 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,136 @@
1+
/*
2+
* Licensed to the Apache Software Foundation (ASF) under one
3+
* or more contributor license agreements. See the NOTICE file
4+
* distributed with this work for additional information
5+
* regarding copyright ownership. The ASF licenses this file
6+
* to you under the Apache License, Version 2.0 (the
7+
* "License"); you may not use this file except in compliance
8+
* with the License. You may obtain a copy of the License at
9+
*
10+
* http://www.apache.org/licenses/LICENSE-2.0
11+
*
12+
* Unless required by applicable law or agreed to in writing,
13+
* software distributed under the License is distributed on an
14+
* "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
15+
* KIND, either express or implied. See the License for the
16+
* specific language governing permissions and limitations
17+
* under the License.
18+
*/
19+
20+
package com.antgroup.geaflow.dsl.udf.table.agg;
21+
22+
import com.antgroup.geaflow.dsl.common.exception.GeaFlowDSLException;
23+
import com.antgroup.geaflow.dsl.common.function.Description;
24+
import com.antgroup.geaflow.dsl.common.function.UDAF;
25+
import com.antgroup.geaflow.dsl.common.function.UDAFArguments;
26+
import com.antgroup.geaflow.dsl.udf.table.agg.PercentileDouble.Accumulator;
27+
import com.antgroup.geaflow.dsl.udf.table.agg.PercentileDouble.MultiArguments;
28+
import java.io.Serializable;
29+
import java.math.BigDecimal;
30+
import java.util.ArrayList;
31+
import java.util.List;
32+
import org.apache.commons.math3.stat.StatUtils;
33+
34+
@Description(name = "percentile", description = "percentile agg function for double")
35+
public class PercentileDouble extends UDAF<MultiArguments, Accumulator, Double> {
36+
37+
@Override
38+
public Accumulator createAccumulator() {
39+
return new Accumulator();
40+
}
41+
42+
@Override
43+
public void accumulate(Accumulator accumulator, MultiArguments input) {
44+
if (input != null) {
45+
accumulator.setPercent(PercentileDouble.getPercent(input.getParam(1)));
46+
accumulator.getValueList().add((double) input.getParam(0));
47+
}
48+
}
49+
50+
public static double getPercent(Object percent) {
51+
if (percent instanceof BigDecimal) {
52+
return ((BigDecimal) percent).doubleValue();
53+
} else if (percent instanceof Double) {
54+
return ((Double) percent);
55+
} else if (percent instanceof Float) {
56+
return ((Float) percent);
57+
} else if (percent instanceof Long) {
58+
return ((Long) percent).doubleValue();
59+
} else if (percent instanceof Integer) {
60+
return ((Integer) percent).doubleValue();
61+
}
62+
throw new GeaFlowDSLException("Percentile not support percent type: " + percent);
63+
}
64+
65+
@Override
66+
public void merge(Accumulator accumulator, Iterable<Accumulator> its) {
67+
for (Accumulator it : its) {
68+
if (it != null) {
69+
accumulator.getValueList().addAll(it.getValueList());
70+
accumulator.setPercent(it.getPercent());
71+
}
72+
}
73+
}
74+
75+
@Override
76+
public void resetAccumulator(Accumulator accumulator) {
77+
accumulator.setValueList(new ArrayList<>());
78+
}
79+
80+
@Override
81+
public Double getValue(Accumulator accumulator) {
82+
List<Double> valueList = accumulator.getValueList();
83+
double[] values = new double[valueList.size()];
84+
for (int i = 0; i < valueList.size(); i++) {
85+
values[i] = valueList.get(i);
86+
}
87+
return StatUtils.percentile(values, accumulator.getPercent());
88+
}
89+
90+
public static class Accumulator implements Serializable {
91+
92+
private static final long serialVersionUID = 7024955653427528364L;
93+
94+
private List<Double> valueList;
95+
private double percent;
96+
97+
public Accumulator() {
98+
this.valueList = new ArrayList<>();
99+
}
100+
101+
public Accumulator(double value) {
102+
this.valueList.add(value);
103+
}
104+
105+
public List<Double> getValueList() {
106+
return valueList;
107+
}
108+
109+
public void setValueList(List<Double> valueList) {
110+
this.valueList = valueList;
111+
}
112+
113+
public double getPercent() {
114+
return percent;
115+
}
116+
117+
public void setPercent(double percent) {
118+
this.percent = percent;
119+
}
120+
}
121+
122+
public static class MultiArguments extends UDAFArguments {
123+
124+
public MultiArguments() {
125+
}
126+
127+
@Override
128+
public List<Class<?>> getParamTypes() {
129+
List<Class<?>> types = new ArrayList<>();
130+
types.add(Double.class);
131+
types.add(Double.class);
132+
return types;
133+
}
134+
}
135+
}
136+

‎geaflow/geaflow-dsl/geaflow-dsl-plan/src/main/java/com/antgroup/geaflow/dsl/udf/table/agg/PercentileInteger.java‎

Lines changed: 86 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,86 @@
1+
/*
2+
* Licensed to the Apache Software Foundation (ASF) under one
3+
* or more contributor license agreements. See the NOTICE file
4+
* distributed with this work for additional information
5+
* regarding copyright ownership. The ASF licenses this file
6+
* to you under the Apache License, Version 2.0 (the
7+
* "License"); you may not use this file except in compliance
8+
* with the License. You may obtain a copy of the License at
9+
*
10+
* http://www.apache.org/licenses/LICENSE-2.0
11+
*
12+
* Unless required by applicable law or agreed to in writing,
13+
* software distributed under the License is distributed on an
14+
* "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
15+
* KIND, either express or implied. See the License for the
16+
* specific language governing permissions and limitations
17+
* under the License.
18+
*/
19+
20+
package com.antgroup.geaflow.dsl.udf.table.agg;
21+
22+
import com.antgroup.geaflow.dsl.common.function.Description;
23+
import com.antgroup.geaflow.dsl.common.function.UDAF;
24+
import com.antgroup.geaflow.dsl.common.function.UDAFArguments;
25+
import com.antgroup.geaflow.dsl.udf.table.agg.PercentileDouble.Accumulator;
26+
import com.antgroup.geaflow.dsl.udf.table.agg.PercentileInteger.MultiArguments;
27+
import java.util.ArrayList;
28+
import java.util.List;
29+
import org.apache.commons.math3.stat.StatUtils;
30+
31+
@Description(name = "percentile", description = "percentile agg function for integer")
32+
public class PercentileInteger extends UDAF<MultiArguments, Accumulator, Double> {
33+
34+
@Override
35+
public Accumulator createAccumulator() {
36+
return new Accumulator();
37+
}
38+
39+
@Override
40+
public void accumulate(Accumulator accumulator, MultiArguments input) {
41+
if (input != null) {
42+
accumulator.setPercent(PercentileDouble.getPercent(input.getParam(1)));
43+
accumulator.getValueList().add((double) (int) input.getParam(0));
44+
}
45+
}
46+
47+
@Override
48+
public void merge(Accumulator accumulator, Iterable<Accumulator> its) {
49+
for (Accumulator it : its) {
50+
if (it != null) {
51+
accumulator.getValueList().addAll(it.getValueList());
52+
accumulator.setPercent(it.getPercent());
53+
}
54+
}
55+
}
56+
57+
@Override
58+
public void resetAccumulator(Accumulator accumulator) {
59+
accumulator.setValueList(new ArrayList<>());
60+
}
61+
62+
@Override
63+
public Double getValue(Accumulator accumulator) {
64+
List<Double> valueList = accumulator.getValueList();
65+
double[] values = new double[valueList.size()];
66+
for (int i = 0; i < valueList.size(); i++) {
67+
values[i] = valueList.get(i);
68+
}
69+
return StatUtils.percentile(values, accumulator.getPercent());
70+
}
71+
72+
public static class MultiArguments extends UDAFArguments {
73+
74+
public MultiArguments() {
75+
}
76+
77+
@Override
78+
public List<Class<?>> getParamTypes() {
79+
List<Class<?>> types = new ArrayList<>();
80+
types.add(Integer.class);
81+
types.add(Double.class);
82+
return types;
83+
}
84+
}
85+
}
86+

‎geaflow/geaflow-dsl/geaflow-dsl-plan/src/main/java/com/antgroup/geaflow/dsl/udf/table/agg/PercentileLong.java‎

Lines changed: 86 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,86 @@
1+
/*
2+
* Licensed to the Apache Software Foundation (ASF) under one
3+
* or more contributor license agreements. See the NOTICE file
4+
* distributed with this work for additional information
5+
* regarding copyright ownership. The ASF licenses this file
6+
* to you under the Apache License, Version 2.0 (the
7+
* "License"); you may not use this file except in compliance
8+
* with the License. You may obtain a copy of the License at
9+
*
10+
* http://www.apache.org/licenses/LICENSE-2.0
11+
*
12+
* Unless required by applicable law or agreed to in writing,
13+
* software distributed under the License is distributed on an
14+
* "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
15+
* KIND, either express or implied. See the License for the
16+
* specific language governing permissions and limitations
17+
* under the License.
18+
*/
19+
20+
package com.antgroup.geaflow.dsl.udf.table.agg;
21+
22+
import com.antgroup.geaflow.dsl.common.function.Description;
23+
import com.antgroup.geaflow.dsl.common.function.UDAF;
24+
import com.antgroup.geaflow.dsl.common.function.UDAFArguments;
25+
import com.antgroup.geaflow.dsl.udf.table.agg.PercentileDouble.Accumulator;
26+
import com.antgroup.geaflow.dsl.udf.table.agg.PercentileLong.MultiArguments;
27+
import java.util.ArrayList;
28+
import java.util.List;
29+
import org.apache.commons.math3.stat.StatUtils;
30+
31+
@Description(name = "percentile", description = "percentile agg function for long")
32+
public class PercentileLong extends UDAF<MultiArguments, Accumulator, Double> {
33+
34+
@Override
35+
public Accumulator createAccumulator() {
36+
return new Accumulator();
37+
}
38+
39+
@Override
40+
public void accumulate(Accumulator accumulator, MultiArguments input) {
41+
if (input != null) {
42+
accumulator.setPercent(PercentileDouble.getPercent(input.getParam(1)));
43+
accumulator.getValueList().add((double) (long) input.getParam(0));
44+
}
45+
}
46+
47+
@Override
48+
public void merge(Accumulator accumulator, Iterable<Accumulator> its) {
49+
for (Accumulator it : its) {
50+
if (it != null) {
51+
accumulator.getValueList().addAll(it.getValueList());
52+
accumulator.setPercent(it.getPercent());
53+
}
54+
}
55+
}
56+
57+
@Override
58+
public void resetAccumulator(Accumulator accumulator) {
59+
accumulator.setValueList(new ArrayList<>());
60+
}
61+
62+
@Override
63+
public Double getValue(Accumulator accumulator) {
64+
List<Double> valueList = accumulator.getValueList();
65+
double[] values = new double[valueList.size()];
66+
for (int i = 0; i < valueList.size(); i++) {
67+
values[i] = valueList.get(i);
68+
}
69+
return StatUtils.percentile(values, accumulator.getPercent());
70+
}
71+
72+
public static class MultiArguments extends UDAFArguments {
73+
74+
public MultiArguments() {
75+
}
76+
77+
@Override
78+
public List<Class<?>> getParamTypes() {
79+
List<Class<?>> types = new ArrayList<>();
80+
types.add(Long.class);
81+
types.add(Double.class);
82+
return types;
83+
}
84+
}
85+
}
86+

‎geaflow/geaflow-dsl/geaflow-dsl-runtime/src/main/java/com/antgroup/geaflow/dsl/runtime/plan/PhysicAggregateRelNode.java‎

Lines changed: 11 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -52,6 +52,9 @@
5252
import com.antgroup.geaflow.dsl.udf.table.agg.MinDouble;
5353
import com.antgroup.geaflow.dsl.udf.table.agg.MinInteger;
5454
import com.antgroup.geaflow.dsl.udf.table.agg.MinLong;
55+
import com.antgroup.geaflow.dsl.udf.table.agg.PercentileDouble;
56+
import com.antgroup.geaflow.dsl.udf.table.agg.PercentileInteger;
57+
import com.antgroup.geaflow.dsl.udf.table.agg.PercentileLong;
5558
import com.antgroup.geaflow.dsl.udf.table.agg.StdDevSampDouble;
5659
import com.antgroup.geaflow.dsl.udf.table.agg.StdDevSampInteger;
5760
import com.antgroup.geaflow.dsl.udf.table.agg.StdDevSampLong;
@@ -97,6 +100,9 @@ public class PhysicAggregateRelNode extends Aggregate implements PhysicRelNode<R
97100

98101
public static final String UDAF_STDDEV_SAMP = "STDDEV_SAMP";
99102

103+
public static final String UDAF_PERCENTILE = "PERCENTILE";
104+
105+
100106
public PhysicAggregateRelNode(
101107
RelOptCluster cluster,
102108
RelTraitSet traits,
@@ -445,6 +451,11 @@ private List<AggFunctionCall> buildAggFunctionCalls() {
445451
aggClasses.add(StdDevSampDouble.class);
446452
aggClasses.add(StdDevSampInteger.class);
447453
break;
454+
case UDAF_PERCENTILE:
455+
aggClasses.add(PercentileLong.class);
456+
aggClasses.add(PercentileInteger.class);
457+
aggClasses.add(PercentileDouble.class);
458+
break;
448459
default:
449460
throw new GeaFlowDSLException("Not support aggregate function " + aggName);
450461
}

‎geaflow/geaflow-dsl/geaflow-dsl-runtime/src/test/java/com/antgroup/geaflow/dsl/runtime/query/AggregateTest.java‎

Lines changed: 9 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -122,6 +122,15 @@ public void testAggregate_011() throws Exception {
122122
.checkSinkResult();
123123
}
124124

125+
@Test
126+
public void testAggregate_012() throws Exception {
127+
QueryTester
128+
.build()
129+
.withQueryPath("/query/aggregate_012.sql")
130+
.execute()
131+
.checkSinkResult();
132+
}
133+
125134
@Test
126135
public void testStreamAggregate_001() throws Exception {
127136
QueryTester

‎geaflow/geaflow-dsl/geaflow-dsl-runtime/src/test/resources/expect/aggregate_012.txt‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1 @@
1+
1.0,1.0,15.7

0 commit comments

Comments
 (0)