Skip to content

Commit ce34c17

Browse files
ArmxGArmxG
committed
feat: support degree algorithm (#795)
1 parent f8be722 commit ce34c17

5 files changed

Lines changed: 175 additions & 0 deletions

File tree

‎geaflow/geaflow-dsl/geaflow-dsl-plan/src/main/java/org/apache/geaflow/dsl/schema/function/BuildInSqlFunctionTable.java‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -39,6 +39,7 @@
3939
import org.apache.geaflow.dsl.udf.graph.ClusterCoefficient;
4040
import org.apache.geaflow.dsl.udf.graph.CommonNeighbors;
4141
import org.apache.geaflow.dsl.udf.graph.ConnectedComponents;
42+
import org.apache.geaflow.dsl.udf.graph.Degree;
4243
import org.apache.geaflow.dsl.udf.graph.IncKHopAlgorithm;
4344
import org.apache.geaflow.dsl.udf.graph.IncMinimumSpanningTree;
4445
import org.apache.geaflow.dsl.udf.graph.IncWeakConnectedComponents;
@@ -228,6 +229,7 @@ public class BuildInSqlFunctionTable extends ListSqlOperatorTable {
228229
.add(GeaFlowFunction.of(PageRank.class))
229230
.add(GeaFlowFunction.of(KHop.class))
230231
.add(GeaFlowFunction.of(KCore.class))
232+
.add(GeaFlowFunction.of(Degree.class))
231233
.add(GeaFlowFunction.of(IncrementalKCore.class))
232234
.add(GeaFlowFunction.of(IncMinimumSpanningTree.class))
233235
.add(GeaFlowFunction.of(ClosenessCentrality.class))

‎geaflow/geaflow-dsl/geaflow-dsl-plan/src/main/java/org/apache/geaflow/dsl/udf/graph/Degree.java‎

Lines changed: 79 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,79 @@
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 org.apache.geaflow.dsl.udf.graph;
21+
22+
import java.util.Iterator;
23+
import java.util.Optional;
24+
import org.apache.geaflow.common.type.primitive.IntegerType;
25+
import org.apache.geaflow.dsl.common.algo.AlgorithmRuntimeContext;
26+
import org.apache.geaflow.dsl.common.algo.AlgorithmUserFunction;
27+
import org.apache.geaflow.dsl.common.data.Row;
28+
import org.apache.geaflow.dsl.common.data.RowVertex;
29+
import org.apache.geaflow.dsl.common.data.impl.ObjectRow;
30+
import org.apache.geaflow.dsl.common.function.Description;
31+
import org.apache.geaflow.dsl.common.types.GraphSchema;
32+
import org.apache.geaflow.dsl.common.types.StructType;
33+
import org.apache.geaflow.dsl.common.types.TableField;
34+
import org.apache.geaflow.model.graph.edge.EdgeDirection;
35+
36+
@Description(name = "degree", description = "built-in udga for Degree")
37+
public class Degree implements AlgorithmUserFunction<Object, Integer> {
38+
39+
private AlgorithmRuntimeContext<Object, Integer> context;
40+
41+
@Override
42+
public void init(AlgorithmRuntimeContext<Object, Integer> context, Object[] params) {
43+
this.context = context;
44+
if (params.length > 0) {
45+
throw new IllegalArgumentException(
46+
"The degree algorithm takes no arguments, usage: degree()");
47+
}
48+
}
49+
50+
@Override
51+
public void process(RowVertex vertex, Optional<Row> updatedValues, Iterator<Integer> messages) {
52+
updatedValues.ifPresent(vertex::setValue);
53+
// Degree needs no message passing: each vertex can count its own
54+
// adjacent edges directly in the first (and only) iteration.
55+
if (context.getCurrentIterationId() == 1L) {
56+
int inDegree = context.loadEdges(EdgeDirection.IN).size();
57+
int outDegree = context.loadEdges(EdgeDirection.OUT).size();
58+
context.updateVertexValue(ObjectRow.create(inDegree, outDegree));
59+
}
60+
}
61+
62+
@Override
63+
public void finish(RowVertex graphVertex, Optional<Row> updatedValues) {
64+
updatedValues.ifPresent(graphVertex::setValue);
65+
int inDegree = (int) graphVertex.getValue().getField(0, IntegerType.INSTANCE);
66+
int outDegree = (int) graphVertex.getValue().getField(1, IntegerType.INSTANCE);
67+
context.take(ObjectRow.create(graphVertex.getId(), inDegree, outDegree));
68+
}
69+
70+
@Override
71+
public StructType getOutputType(GraphSchema graphSchema) {
72+
return new StructType(
73+
new TableField("id", graphSchema.getIdType(), false),
74+
new TableField("in_degree", IntegerType.INSTANCE, false),
75+
new TableField("out_degree", IntegerType.INSTANCE, false)
76+
);
77+
}
78+
79+
}

‎geaflow/geaflow-dsl/geaflow-dsl-runtime/src/test/java/org/apache/geaflow/dsl/runtime/query/GQLAlgorithmTest.java‎

Lines changed: 9 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -115,6 +115,15 @@ public void testAlgorithmKCore() throws Exception {
115115
.checkSinkResult();
116116
}
117117

118+
@Test
119+
public void testAlgorithmDegree() throws Exception {
120+
QueryTester
121+
.build()
122+
.withQueryPath("/query/gql_algorithm_degree.sql")
123+
.execute()
124+
.checkSinkResult();
125+
}
126+
118127
@Test
119128
public void testAlgorithmClosenessCentrality() throws Exception {
120129
QueryTester

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

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,6 @@
1+
1,1,3
2+
2,0,1
3+
3,2,1
4+
4,2,2
5+
5,1,2
6+
6,3,0

‎geaflow/geaflow-dsl/geaflow-dsl-runtime/src/test/resources/query/gql_algorithm_degree.sql‎

Lines changed: 79 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,79 @@
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+
set geaflow.dsl.window.size = -1;
21+
set geaflow.dsl.ignore.exception = true;
22+
23+
CREATE GRAPH IF NOT EXISTS g4 (
24+
Vertex v4 (
25+
vid varchar ID,
26+
vvalue int
27+
),
28+
Edge e4 (
29+
srcId varchar SOURCE ID,
30+
targetId varchar DESTINATION ID
31+
)
32+
) WITH (
33+
storeType='rocksdb',
34+
shardCount = 1
35+
);
36+
37+
CREATE TABLE IF NOT EXISTS v_source (
38+
v_id varchar,
39+
v_value int,
40+
ts varchar,
41+
type varchar
42+
) WITH (
43+
type='file',
44+
geaflow.dsl.file.path = 'resource:///input/test_vertex'
45+
);
46+
47+
CREATE TABLE IF NOT EXISTS e_source (
48+
src_id varchar,
49+
dst_id varchar
50+
) WITH (
51+
type='file',
52+
geaflow.dsl.file.path = 'resource:///input/test_edge'
53+
);
54+
55+
CREATE TABLE IF NOT EXISTS tbl_result (
56+
v_id varchar,
57+
in_degree int,
58+
out_degree int
59+
) WITH (
60+
type='file',
61+
geaflow.dsl.file.path = '${target}'
62+
);
63+
64+
USE GRAPH g4;
65+
66+
INSERT INTO g4.v4(vid, vvalue)
67+
SELECT
68+
v_id, v_value
69+
FROM v_source;
70+
71+
INSERT INTO g4.e4(srcId, targetId)
72+
SELECT
73+
src_id, dst_id
74+
FROM e_source;
75+
76+
INSERT INTO tbl_result(v_id, in_degree, out_degree)
77+
CALL degree() YIELD (vid, inDegree, outDegree)
78+
RETURN vid, inDegree, outDegree
79+
;

0 commit comments

Comments
 (0)