kitalkuyo-gita commented on code in PR #670: URL: https://github.com/apache/geaflow/pull/670#discussion_r2544246576
########## geaflow/geaflow-dsl/geaflow-dsl-plan/src/main/java/org/apache/geaflow/dsl/udf/graph/ConnectedComponents.java: ########## @@ -0,0 +1,110 @@ +/* + * 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 outputKeyName = "component"; + private int iteration = 20; + + @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, [outputKeyName]])"); + } + if (parameters.length > 0) { + iteration = Integer.parseInt(String.valueOf(parameters[0])); + } + if (parameters.length > 1) { + outputKeyName = 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) { + String initValue = String.valueOf(vertex.getId()); + sendMessageToNeighbors(edges, initValue); + context.sendMessage(vertex.getId(), initValue); + context.updateVertexValue(ObjectRow.create(initValue)); + } else if (context.getCurrentIterationId() < iteration) { + String minComponent = messages.next(); + while (messages.hasNext()) { Review Comment: Initializing the label to a string of vertex.id is common practice. Retaining the current label when messages are empty is reasonable, but the `process` method lacks protection for `messages.hasNext()`. Although the code uses `while(messages.hasNext())`, it doesn't check at the beginning (the problem is more serious with `ConnectedComponents`). `LabelPropagation` is safe here (because the `while` loop is skipped), but implementation consistency still needs attention. ########## 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: Copying the results of `context.loadEdges(...)` to an `ArrayList(List<RowEdge> edges = new ArrayList<>(...))` can consume a lot of memory and time, especially with large datasets. Consider using the `stream` API for this purpose. ########## 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 outputKeyName = "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, [outputKeyName]])"); + } + if (parameters.length > 0) { + iteration = Integer.parseInt(String.valueOf(parameters[0])); + } + if (parameters.length > 1) { + outputKeyName = 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); + + while (messages.hasNext()) { + String label = messages.next(); + labelCounts.put(label, labelCounts.getOrDefault(label, 0) + 1); + } + + String mostFrequentLabel = currentLabel; + int maxCount = 0; + + for (Map.Entry<String, Integer> entry : labelCounts.entrySet()) { + String label = entry.getKey(); + int count = entry.getValue(); + if (count >= maxCount && label.compareTo(mostFrequentLabel) < 0) { + mostFrequentLabel = label; Review Comment: After counting neighbor labels, the condition for selecting mostFrequentLabel is `if (count >= maxCount && label.compareTo(mostFrequentLabel) < 0)`. This condition combination has two points that need review: - 1. Using `>=` instead of `>`: When `count == maxCount`, lexicographic compare is considered, but only when `label < mostFrequentLabel` (lexicographically smaller) will a replacement be made. This biases towards selecting labels with smaller lexicographical order as tie-breaks. While the logic still works when `maxCount` is initially 0 and `mostFrequentLabel` is initially `currentLabel`, semantically, the tie-break strategy should be explicit (random, largest ID, or smallest ID). The current approach is arbitrary but deterministic. - 2. When `labelCounts` is empty (no message received), `mostFrequentLabel` retains `currentLabel` (which is reasonable). Recommendation: Explicitly define and comment out the tie-break strategy. Common options: - 1. Random selection (requires pseudo-random numbers and may affect repeatability); - 2. Select the label with the smallest/largest id (if the label is based on the vertex id) — this is deterministic and easily interpretable; - 3. Keep the current label (do not change it if there is no absolute majority) — this can reduce jitter. For better repeatability and interpretability, it is recommended to use numerical comparison (if the label can be parsed into a long integer id); otherwise, use lexicographical order and comment it out as a design decision. ########## 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 outputKeyName = "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, [outputKeyName]])"); + } + if (parameters.length > 0) { + iteration = Integer.parseInt(String.valueOf(parameters[0])); + } + if (parameters.length > 1) { + outputKeyName = 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)); Review Comment: The way neighbor IDs are sent might be incorrect (causing spontaneous or duplicate sending across all edges). Load edges: `context.loadEdges(EdgeDirection.BOTH)`, then iterate through the edges object `rowEdge.getTargetId()` and call `context.sendMessage(...)`. For `BOTH`, the framework returns edges in both directions (source->target and target->source). Directly retrieving `targetId` might send the message back to itself (if `targetId == currentId`) or duplicate the same neighbor twice (if they were originally represented by each other). Recommendation: The "other end" endpoint ID should be explicitly retrieved (see the recommendations for understanding ConnectedComponents). -- 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]
