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..239f7e1ef 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 @@ -35,6 +35,7 @@ import org.apache.geaflow.dsl.planner.GQLJavaTypeFactory; import org.apache.geaflow.dsl.schema.GeaFlowFunction; import org.apache.geaflow.dsl.udf.graph.AllSourceShortestPath; +import org.apache.geaflow.dsl.udf.graph.BetweennessCentrality; import org.apache.geaflow.dsl.udf.graph.ClosenessCentrality; import org.apache.geaflow.dsl.udf.graph.ClusterCoefficient; import org.apache.geaflow.dsl.udf.graph.CommonNeighbors; @@ -231,6 +232,7 @@ public class BuildInSqlFunctionTable extends ListSqlOperatorTable { .add(GeaFlowFunction.of(IncrementalKCore.class)) .add(GeaFlowFunction.of(IncMinimumSpanningTree.class)) .add(GeaFlowFunction.of(ClosenessCentrality.class)) + .add(GeaFlowFunction.of(BetweennessCentrality.class)) .add(GeaFlowFunction.of(WeakConnectedComponents.class)) .add(GeaFlowFunction.of(TriangleCount.class)) .add(GeaFlowFunction.of(ClusterCoefficient.class)) diff --git a/geaflow/geaflow-dsl/geaflow-dsl-plan/src/main/java/org/apache/geaflow/dsl/udf/graph/BetweennessCentrality.java b/geaflow/geaflow-dsl/geaflow-dsl-plan/src/main/java/org/apache/geaflow/dsl/udf/graph/BetweennessCentrality.java new file mode 100644 index 000000000..c712bdb24 --- /dev/null +++ b/geaflow/geaflow-dsl/geaflow-dsl-plan/src/main/java/org/apache/geaflow/dsl/udf/graph/BetweennessCentrality.java @@ -0,0 +1,270 @@ +/* + * 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.ArrayList; +import java.util.HashMap; +import java.util.Iterator; +import java.util.List; +import java.util.Map; +import java.util.Optional; +import org.apache.geaflow.common.type.primitive.DoubleType; +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.RowEdge; +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; + +/** + * Built-in UDGA computing Betweenness Centrality via a vertex-centric + * adaptation of Brandes' algorithm. + * + *

All vertices act as sources simultaneously. Each vertex keeps per-source + * state {@code st[source] = [dist, sigma, delta, childCount, respCount, + * recvChild, pendingSum, forwarded, backwardDone]} plus its per-source + * predecessor list.

+ * + *

Phase 1 (forward): a BFS from every source counts the number of shortest + * paths ({@code sigma}) reaching each vertex and records predecessors. Every + * forwarded neighbour replies ACK (it is a child) or NACK (it is not), so a + * vertex learns how many children it has without a global barrier.

+ * + *

Phase 2 (backward): once a vertex has heard from all children for a source + * it finalizes its dependency {@code delta = sigma * sum((1 + delta_child) / + * sigma_child)}, adds it to its betweenness score, and pushes the contribution + * to its predecessors.

+ * + *

Directed, unnormalized betweenness (endpoints excluded). Complexity is + * O(V * E) time; space is O(V^2) because per-source state is kept on every + * vertex, so this suits small / medium graphs.

+ */ +@Description(name = "betweenness_centrality", description = "built-in udga for BetweennessCentrality") +public class BetweennessCentrality implements AlgorithmUserFunction> { + + // Message tags. + private static final int FWD = 0; + private static final int ACK = 1; + private static final int NACK = 2; + private static final int BWD = 3; + + // Per-source state array indices. + private static final int DIST = 0; + private static final int SIGMA = 1; + private static final int DELTA = 2; + private static final int CHILD_COUNT = 3; + private static final int RESP_COUNT = 4; + private static final int RECV_CHILD = 5; + private static final int PENDING_SUM = 6; + private static final int FORWARDED = 7; + private static final int BACKWARD_DONE = 8; + private static final int STATE_LEN = 9; + + private AlgorithmRuntimeContext> context; + + @Override + public void init(AlgorithmRuntimeContext> context, Object[] params) { + this.context = context; + if (params.length > 0) { + throw new IllegalArgumentException( + "The betweenness_centrality algorithm takes no arguments, usage: betweenness_centrality()"); + } + } + + @Override + @SuppressWarnings("unchecked") + public void process(RowVertex vertex, Optional updatedValues, Iterator> messages) { + Object vid = vertex.getId(); + List outEdges = context.loadEdges(EdgeDirection.OUT); + int outDegree = outEdges.size(); + + // Iteration 1: every vertex initializes itself as a source and scatters. + if (context.getCurrentIterationId() == 1L) { + double[] arr = new double[STATE_LEN]; + arr[DIST] = 0; + arr[SIGMA] = 1; + arr[FORWARDED] = 1; + Map state = newState(); + Map st = (Map) state.get("st"); + st.put(vid, arr); + Map> preds = (Map>) state.get("preds"); + preds.put(vid, new ArrayList<>()); + for (RowEdge edge : outEdges) { + context.sendMessage(edge.getTargetId(), forwardMsg(vid, vid, 0L, 1.0)); + } + context.updateVertexValue(ObjectRow.create(state)); + return; + } + + Map state = (Map) updatedValues.get().getField(0, null); + double bc = (Double) state.get("bc"); + final Map st = (Map) state.get("st"); + + // Group incoming forward messages by source; apply ack/nack/backward directly. + Map>> forwardBySource = new HashMap<>(); + while (messages.hasNext()) { + List msg = messages.next(); + int tag = ((Number) msg.get(0)).intValue(); + Object source = msg.get(1); + if (tag == FWD) { + forwardBySource.computeIfAbsent(source, k -> new ArrayList<>()).add(msg); + } else if (tag == ACK) { + double[] arr = st.get(source); + if (arr != null) { + arr[CHILD_COUNT]++; + arr[RESP_COUNT]++; + } + } else if (tag == NACK) { + double[] arr = st.get(source); + if (arr != null) { + arr[RESP_COUNT]++; + } + } else if (tag == BWD) { + double value = ((Number) msg.get(2)).doubleValue(); + double[] arr = st.get(source); + if (arr != null) { + arr[PENDING_SUM] += value; + arr[RECV_CHILD]++; + } + } + } + + // Handle forward messages: update sigma / predecessors, reply ack/nack, scatter once. + final Map> preds = (Map>) state.get("preds"); + for (Map.Entry>> entry : forwardBySource.entrySet()) { + Object source = entry.getKey(); + for (List msg : entry.getValue()) { + Object sender = msg.get(2); + int senderDist = ((Number) msg.get(3)).intValue(); + double senderSigma = ((Number) msg.get(4)).doubleValue(); + int candidate = senderDist + 1; + double[] arr = st.get(source); + if (arr == null) { + arr = new double[STATE_LEN]; + arr[DIST] = candidate; + arr[SIGMA] = 0; + st.put(source, arr); + preds.put(source, new ArrayList<>()); + } + if ((int) arr[DIST] == candidate) { + arr[SIGMA] += senderSigma; + preds.get(source).add(sender); + context.sendMessage(sender, tagSourceMsg(ACK, source)); + } else { + context.sendMessage(sender, tagSourceMsg(NACK, source)); + } + } + double[] arr = st.get(source); + if (arr[FORWARDED] == 0) { + arr[FORWARDED] = 1; + long dist = (long) arr[DIST]; + double sigma = arr[SIGMA]; + for (RowEdge edge : outEdges) { + context.sendMessage(edge.getTargetId(), forwardMsg(source, vid, dist, sigma)); + } + } + } + + // Backward accumulation: finalize any source whose children have all reported. + for (Object source : new ArrayList<>(st.keySet())) { + double[] arr = st.get(source); + if (arr[BACKWARD_DONE] == 1) { + continue; + } + boolean forwarded = arr[FORWARDED] == 1; + boolean childCountFinal = forwarded && arr[RESP_COUNT] >= outDegree; + if (childCountFinal && arr[RECV_CHILD] >= arr[CHILD_COUNT]) { + double sigma = arr[SIGMA]; + double delta = sigma * arr[PENDING_SUM]; + arr[DELTA] = delta; + if (!source.equals(vid)) { + bc += delta; + } + arr[BACKWARD_DONE] = 1; + double value = (1.0 + delta) / sigma; + for (Object pred : preds.get(source)) { + context.sendMessage(pred, backwardMsg(source, value)); + } + } + } + + state.put("bc", bc); + context.updateVertexValue(ObjectRow.create(state)); + } + + @Override + @SuppressWarnings("unchecked") + public void finish(RowVertex vertex, Optional updatedValues) { + double bc = 0.0; + if (updatedValues.isPresent()) { + Map state = (Map) updatedValues.get().getField(0, null); + if (state != null && state.get("bc") != null) { + bc = (Double) state.get("bc"); + } + } + context.take(ObjectRow.create(vertex.getId(), bc)); + } + + @Override + public StructType getOutputType(GraphSchema graphSchema) { + return new StructType( + new TableField("id", graphSchema.getIdType(), false), + new TableField("betweenness", DoubleType.INSTANCE, false) + ); + } + + private static Map newState() { + Map state = new HashMap<>(); + state.put("st", new HashMap()); + state.put("preds", new HashMap>()); + state.put("bc", 0.0); + return state; + } + + private static List forwardMsg(Object source, Object sender, long dist, double sigma) { + List msg = new ArrayList<>(5); + msg.add(FWD); + msg.add(source); + msg.add(sender); + msg.add(dist); + msg.add(sigma); + return msg; + } + + private static List tagSourceMsg(int tag, Object source) { + List msg = new ArrayList<>(2); + msg.add(tag); + msg.add(source); + return msg; + } + + private static List backwardMsg(Object source, double value) { + List msg = new ArrayList<>(3); + msg.add(BWD); + msg.add(source); + msg.add(value); + return msg; + } +} 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..e31fa46f7 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 @@ -124,6 +124,15 @@ public void testAlgorithmClosenessCentrality() throws Exception { .checkSinkResult(); } + @Test + public void testAlgorithmBetweennessCentrality() throws Exception { + QueryTester + .build() + .withQueryPath("/query/gql_algorithm_betweenness.sql") + .execute() + .checkSinkResult(); + } + @Test public void testAlgorithmWeakConnectedComponents() throws Exception { QueryTester diff --git a/geaflow/geaflow-dsl/geaflow-dsl-runtime/src/test/resources/expect/gql_algorithm_betweenness.txt b/geaflow/geaflow-dsl/geaflow-dsl-runtime/src/test/resources/expect/gql_algorithm_betweenness.txt new file mode 100644 index 000000000..462ed4678 --- /dev/null +++ b/geaflow/geaflow-dsl/geaflow-dsl-runtime/src/test/resources/expect/gql_algorithm_betweenness.txt @@ -0,0 +1,6 @@ +1,5.0 +2,0.0 +3,4.5 +4,8.0 +5,0.5 +6,0.0 diff --git a/geaflow/geaflow-dsl/geaflow-dsl-runtime/src/test/resources/query/gql_algorithm_betweenness.sql b/geaflow/geaflow-dsl/geaflow-dsl-runtime/src/test/resources/query/gql_algorithm_betweenness.sql new file mode 100644 index 000000000..031499646 --- /dev/null +++ b/geaflow/geaflow-dsl/geaflow-dsl-runtime/src/test/resources/query/gql_algorithm_betweenness.sql @@ -0,0 +1,78 @@ +/* + * 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, + betweenness double +) 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, betweenness) +CALL betweenness_centrality() YIELD (vid, betweenness) +RETURN vid, betweenness +;