diff --git a/geaflow/geaflow-dsl/geaflow-dsl-plan/src/main/java/org/apache/geaflow/dsl/schema/function/BuildInSqlFunctionTable.java b/geaflow/geaflow-dsl/geaflow-dsl-plan/src/main/java/org/apache/geaflow/dsl/schema/function/BuildInSqlFunctionTable.java index 47addc84a..9a515f0e4 100644 --- a/geaflow/geaflow-dsl/geaflow-dsl-plan/src/main/java/org/apache/geaflow/dsl/schema/function/BuildInSqlFunctionTable.java +++ b/geaflow/geaflow-dsl/geaflow-dsl-plan/src/main/java/org/apache/geaflow/dsl/schema/function/BuildInSqlFunctionTable.java @@ -39,6 +39,7 @@ import org.apache.geaflow.dsl.udf.graph.ClusterCoefficient; import org.apache.geaflow.dsl.udf.graph.CommonNeighbors; import org.apache.geaflow.dsl.udf.graph.ConnectedComponents; +import org.apache.geaflow.dsl.udf.graph.Degree; import org.apache.geaflow.dsl.udf.graph.IncKHopAlgorithm; import org.apache.geaflow.dsl.udf.graph.IncMinimumSpanningTree; import org.apache.geaflow.dsl.udf.graph.IncWeakConnectedComponents; @@ -228,6 +229,7 @@ public class BuildInSqlFunctionTable extends ListSqlOperatorTable { .add(GeaFlowFunction.of(PageRank.class)) .add(GeaFlowFunction.of(KHop.class)) .add(GeaFlowFunction.of(KCore.class)) + .add(GeaFlowFunction.of(Degree.class)) .add(GeaFlowFunction.of(IncrementalKCore.class)) .add(GeaFlowFunction.of(IncMinimumSpanningTree.class)) .add(GeaFlowFunction.of(ClosenessCentrality.class)) diff --git a/geaflow/geaflow-dsl/geaflow-dsl-plan/src/main/java/org/apache/geaflow/dsl/udf/graph/Degree.java b/geaflow/geaflow-dsl/geaflow-dsl-plan/src/main/java/org/apache/geaflow/dsl/udf/graph/Degree.java new file mode 100644 index 000000000..678f3943a --- /dev/null +++ b/geaflow/geaflow-dsl/geaflow-dsl-plan/src/main/java/org/apache/geaflow/dsl/udf/graph/Degree.java @@ -0,0 +1,79 @@ +/* + * 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. + */ + +package org.apache.geaflow.dsl.udf.graph; + +import java.util.Iterator; +import java.util.Optional; +import org.apache.geaflow.common.type.primitive.IntegerType; +import org.apache.geaflow.dsl.common.algo.AlgorithmRuntimeContext; +import org.apache.geaflow.dsl.common.algo.AlgorithmUserFunction; +import org.apache.geaflow.dsl.common.data.Row; +import org.apache.geaflow.dsl.common.data.RowVertex; +import org.apache.geaflow.dsl.common.data.impl.ObjectRow; +import org.apache.geaflow.dsl.common.function.Description; +import org.apache.geaflow.dsl.common.types.GraphSchema; +import org.apache.geaflow.dsl.common.types.StructType; +import org.apache.geaflow.dsl.common.types.TableField; +import org.apache.geaflow.model.graph.edge.EdgeDirection; + +@Description(name = "degree", description = "built-in udga for Degree") +public class Degree implements AlgorithmUserFunction { + + private AlgorithmRuntimeContext context; + + @Override + public void init(AlgorithmRuntimeContext context, Object[] params) { + this.context = context; + if (params.length > 0) { + throw new IllegalArgumentException( + "The degree algorithm takes no arguments, usage: degree()"); + } + } + + @Override + public void process(RowVertex vertex, Optional updatedValues, Iterator messages) { + updatedValues.ifPresent(vertex::setValue); + // Degree needs no message passing: each vertex can count its own + // adjacent edges directly in the first (and only) iteration. + if (context.getCurrentIterationId() == 1L) { + int inDegree = context.loadEdges(EdgeDirection.IN).size(); + int outDegree = context.loadEdges(EdgeDirection.OUT).size(); + context.updateVertexValue(ObjectRow.create(inDegree, outDegree)); + } + } + + @Override + public void finish(RowVertex graphVertex, Optional updatedValues) { + updatedValues.ifPresent(graphVertex::setValue); + int inDegree = (int) graphVertex.getValue().getField(0, IntegerType.INSTANCE); + int outDegree = (int) graphVertex.getValue().getField(1, IntegerType.INSTANCE); + context.take(ObjectRow.create(graphVertex.getId(), inDegree, outDegree)); + } + + @Override + public StructType getOutputType(GraphSchema graphSchema) { + return new StructType( + new TableField("id", graphSchema.getIdType(), false), + new TableField("in_degree", IntegerType.INSTANCE, false), + new TableField("out_degree", IntegerType.INSTANCE, false) + ); + } + +} diff --git a/geaflow/geaflow-dsl/geaflow-dsl-runtime/src/test/java/org/apache/geaflow/dsl/runtime/query/GQLAlgorithmTest.java b/geaflow/geaflow-dsl/geaflow-dsl-runtime/src/test/java/org/apache/geaflow/dsl/runtime/query/GQLAlgorithmTest.java index 8c75d6e78..e75b72547 100644 --- a/geaflow/geaflow-dsl/geaflow-dsl-runtime/src/test/java/org/apache/geaflow/dsl/runtime/query/GQLAlgorithmTest.java +++ b/geaflow/geaflow-dsl/geaflow-dsl-runtime/src/test/java/org/apache/geaflow/dsl/runtime/query/GQLAlgorithmTest.java @@ -115,6 +115,15 @@ public void testAlgorithmKCore() throws Exception { .checkSinkResult(); } + @Test + public void testAlgorithmDegree() throws Exception { + QueryTester + .build() + .withQueryPath("/query/gql_algorithm_degree.sql") + .execute() + .checkSinkResult(); + } + @Test public void testAlgorithmClosenessCentrality() throws Exception { QueryTester diff --git a/geaflow/geaflow-dsl/geaflow-dsl-runtime/src/test/resources/expect/gql_algorithm_degree.txt b/geaflow/geaflow-dsl/geaflow-dsl-runtime/src/test/resources/expect/gql_algorithm_degree.txt new file mode 100644 index 000000000..4d8394712 --- /dev/null +++ b/geaflow/geaflow-dsl/geaflow-dsl-runtime/src/test/resources/expect/gql_algorithm_degree.txt @@ -0,0 +1,6 @@ +1,1,3 +2,0,1 +3,2,1 +4,2,2 +5,1,2 +6,3,0 diff --git a/geaflow/geaflow-dsl/geaflow-dsl-runtime/src/test/resources/query/gql_algorithm_degree.sql b/geaflow/geaflow-dsl/geaflow-dsl-runtime/src/test/resources/query/gql_algorithm_degree.sql new file mode 100644 index 000000000..25d02f96e --- /dev/null +++ b/geaflow/geaflow-dsl/geaflow-dsl-runtime/src/test/resources/query/gql_algorithm_degree.sql @@ -0,0 +1,79 @@ +/* + * 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. + */ + +set geaflow.dsl.window.size = -1; +set geaflow.dsl.ignore.exception = true; + +CREATE GRAPH IF NOT EXISTS g4 ( + Vertex v4 ( + vid varchar ID, + vvalue int + ), + Edge e4 ( + srcId varchar SOURCE ID, + targetId varchar DESTINATION ID + ) +) WITH ( + storeType='rocksdb', + shardCount = 1 +); + +CREATE TABLE IF NOT EXISTS v_source ( + v_id varchar, + v_value int, + ts varchar, + type varchar +) WITH ( + type='file', + geaflow.dsl.file.path = 'resource:///input/test_vertex' +); + +CREATE TABLE IF NOT EXISTS e_source ( + src_id varchar, + dst_id varchar +) WITH ( + type='file', + geaflow.dsl.file.path = 'resource:///input/test_edge' +); + +CREATE TABLE IF NOT EXISTS tbl_result ( + v_id varchar, + in_degree int, + out_degree int +) WITH ( + type='file', + geaflow.dsl.file.path = '${target}' +); + +USE GRAPH g4; + +INSERT INTO g4.v4(vid, vvalue) +SELECT +v_id, v_value +FROM v_source; + +INSERT INTO g4.e4(srcId, targetId) +SELECT + src_id, dst_id +FROM e_source; + +INSERT INTO tbl_result(v_id, in_degree, out_degree) +CALL degree() YIELD (vid, inDegree, outDegree) +RETURN vid, inDegree, outDegree +;