Leomrlin commented on code in PR #670: URL: https://github.com/apache/geaflow/pull/670#discussion_r2536114176
########## geaflow/geaflow-dsl/geaflow-dsl-plan/src/main/java/org/apache/geaflow/dsl/udf/graph/ConnectedComponents.java: ########## @@ -0,0 +1,115 @@ +/* + * 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.Iterator; +import java.util.List; +import java.util.Optional; +import org.apache.geaflow.common.type.primitive.StringType; +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; + +@Description(name = "cc", description = "built-in udga for Connected Components Algorithm") +public class ConnectedComponents implements AlgorithmUserFunction<Object, String> { + + private AlgorithmRuntimeContext<Object, String> context; + private String keyFieldName = "component"; + private int iteration = 1000; + + @Override + public void init(AlgorithmRuntimeContext<Object, String> context, Object[] parameters) { + this.context = context; + if (parameters.length > 2) { + throw new IllegalArgumentException( + "Only support zero or more arguments, false arguments " + + "usage: func([iteration, [keyFieldName]])"); + } + if (parameters.length > 0) { + iteration = Integer.parseInt(String.valueOf(parameters[0])); + } + if (parameters.length > 1) { + keyFieldName = String.valueOf(parameters[1]); + } + } + + @Override + public void process(RowVertex vertex, Optional<Row> updatedValues, Iterator<String> messages) { + updatedValues.ifPresent(vertex::setValue); + List<RowEdge> edges = new ArrayList<>(context.loadEdges(EdgeDirection.IN)); + + if (context.getCurrentIterationId() == 1L) { + // Initialize: each vertex starts with its own id as the component id + String initValue = String.valueOf(vertex.getId()); Review Comment: The ID column is not necessarily a string type; you can use its IType. ########## geaflow/geaflow-dsl/geaflow-dsl-plan/src/main/java/org/apache/geaflow/dsl/udf/graph/ConnectedComponents.java: ########## @@ -0,0 +1,115 @@ +/* + * 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.Iterator; +import java.util.List; +import java.util.Optional; +import org.apache.geaflow.common.type.primitive.StringType; +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; + +@Description(name = "cc", description = "built-in udga for Connected Components Algorithm") +public class ConnectedComponents implements AlgorithmUserFunction<Object, String> { + + private AlgorithmRuntimeContext<Object, String> context; + private String keyFieldName = "component"; + private int iteration = 1000; + + @Override + public void init(AlgorithmRuntimeContext<Object, String> context, Object[] parameters) { + this.context = context; + if (parameters.length > 2) { + throw new IllegalArgumentException( + "Only support zero or more arguments, false arguments " + + "usage: func([iteration, [keyFieldName]])"); + } + if (parameters.length > 0) { + iteration = Integer.parseInt(String.valueOf(parameters[0])); + } + if (parameters.length > 1) { + keyFieldName = String.valueOf(parameters[1]); + } + } + + @Override + public void process(RowVertex vertex, Optional<Row> updatedValues, Iterator<String> messages) { + updatedValues.ifPresent(vertex::setValue); + List<RowEdge> edges = new ArrayList<>(context.loadEdges(EdgeDirection.IN)); Review Comment: Propagate only in the IN direction? ########## geaflow/geaflow-dsl/geaflow-dsl-plan/src/main/java/org/apache/geaflow/dsl/udf/graph/ConnectedComponents.java: ########## @@ -0,0 +1,115 @@ +/* + * 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.Iterator; +import java.util.List; +import java.util.Optional; +import org.apache.geaflow.common.type.primitive.StringType; +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; + +@Description(name = "cc", description = "built-in udga for Connected Components Algorithm") +public class ConnectedComponents implements AlgorithmUserFunction<Object, String> { + + private AlgorithmRuntimeContext<Object, String> context; + private String keyFieldName = "component"; + private int iteration = 1000; + + @Override + public void init(AlgorithmRuntimeContext<Object, String> context, Object[] parameters) { + this.context = context; + if (parameters.length > 2) { + throw new IllegalArgumentException( + "Only support zero or more arguments, false arguments " + + "usage: func([iteration, [keyFieldName]])"); Review Comment: This can be easily misunderstood. One might first assume that keyFieldName refers to the column used to compute the cc algo. However, in reality, this parameter specifies the column name for the resulting cc values in the output. It would be helpful to add a clarification note. ########## geaflow/geaflow-dsl/geaflow-dsl-plan/src/main/java/org/apache/geaflow/dsl/udf/graph/ConnectedComponents.java: ########## @@ -0,0 +1,115 @@ +/* + * 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.Iterator; +import java.util.List; +import java.util.Optional; +import org.apache.geaflow.common.type.primitive.StringType; +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; + +@Description(name = "cc", description = "built-in udga for Connected Components Algorithm") +public class ConnectedComponents implements AlgorithmUserFunction<Object, String> { + + private AlgorithmRuntimeContext<Object, String> context; + private String keyFieldName = "component"; + private int iteration = 1000; + + @Override + public void init(AlgorithmRuntimeContext<Object, String> context, Object[] parameters) { + this.context = context; + if (parameters.length > 2) { + throw new IllegalArgumentException( + "Only support zero or more arguments, false arguments " + + "usage: func([iteration, [keyFieldName]])"); + } + if (parameters.length > 0) { + iteration = Integer.parseInt(String.valueOf(parameters[0])); + } + if (parameters.length > 1) { + keyFieldName = String.valueOf(parameters[1]); + } + } + + @Override + public void process(RowVertex vertex, Optional<Row> updatedValues, Iterator<String> messages) { + updatedValues.ifPresent(vertex::setValue); + List<RowEdge> edges = new ArrayList<>(context.loadEdges(EdgeDirection.IN)); + + if (context.getCurrentIterationId() == 1L) { + // Initialize: each vertex starts with its own id as the component id + String initValue = String.valueOf(vertex.getId()); + sendMessageToNeighbors(edges, initValue); + context.sendMessage(vertex.getId(), initValue); + context.updateVertexValue(ObjectRow.create(initValue)); + } else if (context.getCurrentIterationId() < iteration) { + // Find the minimum component id from messages + String minComponent = messages.next(); + while (messages.hasNext()) { + String next = messages.next(); + if (next.compareTo(minComponent) < 0) { + minComponent = next; + } + } + + String currentValue = (String) vertex.getValue().getField(0, StringType.INSTANCE); + // If found smaller component id, update and propagate + if (minComponent.compareTo(currentValue) < 0) { + sendMessageToNeighbors(edges, minComponent); + context.sendMessage(vertex.getId(), minComponent); + context.updateVertexValue(ObjectRow.create(minComponent)); + } else { + context.sendMessage(vertex.getId(), currentValue); Review Comment: In the CC (connected components) algorithm, messages do not need to be sent when vertex values stop updating. Continuously sending messages keeps vertices active and prevents convergence, exhausting the default 1000 iteration limit and reducing efficiency. ########## geaflow/geaflow-dsl/geaflow-dsl-plan/src/main/java/org/apache/geaflow/dsl/udf/graph/LabelPropagation.java: ########## @@ -0,0 +1,121 @@ +/* + * 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.StringType; +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; + +@Description(name = "lpa", description = "built-in udga for Label Propagation Algorithm") +public class LabelPropagation implements AlgorithmUserFunction<Object, String> { + + private AlgorithmRuntimeContext<Object, String> context; + private String keyFieldName = "label"; + private int iteration = 1000; + + @Override + public void init(AlgorithmRuntimeContext<Object, String> context, Object[] parameters) { + this.context = context; + if (parameters.length > 2) { + throw new IllegalArgumentException( + "Only support zero or more arguments, false arguments " + + "usage: func([iteration, [keyFieldName]])"); Review Comment: ditto ########## geaflow/geaflow-dsl/geaflow-dsl-plan/src/main/java/org/apache/geaflow/dsl/udf/graph/ConnectedComponents.java: ########## @@ -0,0 +1,115 @@ +/* + * 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.Iterator; +import java.util.List; +import java.util.Optional; +import org.apache.geaflow.common.type.primitive.StringType; +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; + +@Description(name = "cc", description = "built-in udga for Connected Components Algorithm") +public class ConnectedComponents implements AlgorithmUserFunction<Object, String> { + + private AlgorithmRuntimeContext<Object, String> context; + private String keyFieldName = "component"; + private int iteration = 1000; + + @Override + public void init(AlgorithmRuntimeContext<Object, String> context, Object[] parameters) { + this.context = context; + if (parameters.length > 2) { + throw new IllegalArgumentException( + "Only support zero or more arguments, false arguments " + + "usage: func([iteration, [keyFieldName]])"); + } + if (parameters.length > 0) { + iteration = Integer.parseInt(String.valueOf(parameters[0])); + } + if (parameters.length > 1) { + keyFieldName = String.valueOf(parameters[1]); + } + } + + @Override + public void process(RowVertex vertex, Optional<Row> updatedValues, Iterator<String> messages) { + updatedValues.ifPresent(vertex::setValue); + List<RowEdge> edges = new ArrayList<>(context.loadEdges(EdgeDirection.IN)); + + if (context.getCurrentIterationId() == 1L) { + // Initialize: each vertex starts with its own id as the component id + String initValue = String.valueOf(vertex.getId()); + sendMessageToNeighbors(edges, initValue); + context.sendMessage(vertex.getId(), initValue); + context.updateVertexValue(ObjectRow.create(initValue)); + } else if (context.getCurrentIterationId() < iteration) { + // Find the minimum component id from messages + String minComponent = messages.next(); Review Comment: Is there definitely a message here? ########## geaflow/geaflow-dsl/geaflow-dsl-plan/src/main/java/org/apache/geaflow/dsl/udf/graph/LabelPropagation.java: ########## @@ -0,0 +1,121 @@ +/* + * 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.StringType; +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; + +@Description(name = "lpa", description = "built-in udga for Label Propagation Algorithm") +public class LabelPropagation implements AlgorithmUserFunction<Object, String> { + + private AlgorithmRuntimeContext<Object, String> context; + private String keyFieldName = "label"; + private int iteration = 1000; + + @Override + public void init(AlgorithmRuntimeContext<Object, String> context, Object[] parameters) { + this.context = context; + if (parameters.length > 2) { + throw new IllegalArgumentException( + "Only support zero or more arguments, false arguments " + + "usage: func([iteration, [keyFieldName]])"); + } + if (parameters.length > 0) { + iteration = Integer.parseInt(String.valueOf(parameters[0])); + } + if (parameters.length > 1) { + keyFieldName = String.valueOf(parameters[1]); + } + } + + @Override + public void process(RowVertex vertex, Optional<Row> updatedValues, Iterator<String> messages) { + updatedValues.ifPresent(vertex::setValue); + List<RowEdge> edges = new ArrayList<>(context.loadEdges(EdgeDirection.BOTH)); + + if (context.getCurrentIterationId() == 1L) { + String initLabel = String.valueOf(vertex.getId()); + context.updateVertexValue(ObjectRow.create(initLabel)); + sendMessageToNeighbors(edges, initLabel); + } else if (context.getCurrentIterationId() < iteration) { + Map<String, Integer> labelCounts = new HashMap<>(); + String currentLabel = (String) vertex.getValue().getField(0, StringType.INSTANCE); Review Comment: ditto, The ID column is not necessarily a string type; you can use its IType. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
