[ISSUE-181] Minor fix for dsl traversal

TuGraph-family · Sep 26, 2023 · c56bdb4 · c56bdb4
1 parent 3fd5f64
commit c56bdb4
Show file tree

Hide file tree

Showing 5 changed files with 23 additions and 12 deletions.
diff --git a/...w-dsl/geaflow-dsl-catalog/src/main/java/com/antgroup/geaflow/dsl/schema/GeaFlowGraph.java b/...w-dsl/geaflow-dsl-catalog/src/main/java/com/antgroup/geaflow/dsl/schema/GeaFlowGraph.java
@@ -312,7 +312,7 @@ private void checkFields() {
             if (fieldNames.size() != super.getFields().size()) {
                 throw new GeaFlowDSLException("Duplicate field has found in vertex table: " + getName());
             }
-            if (!fieldNames.contains(idField.toLowerCase())) {
+            if (!fieldNames.contains(idField)) {
                 throw new GeaFlowDSLException("id field:'" + idField + "' is not found in the fields: " + fieldNames);
             }
         }

diff --git a/...sl-runtime/src/main/java/com/antgroup/geaflow/dsl/runtime/traversal/DagTopologyGroup.java b/...sl-runtime/src/main/java/com/antgroup/geaflow/dsl/runtime/traversal/DagTopologyGroup.java
@@ -141,7 +141,7 @@ public int getIterationCount(int currentDepth, StepOperator stepOperator) {
                 maxSubDagIteration = subDagIterationCount;
             }
         }
-        currentDepth += addIteration(currentDepth, maxSubDagIteration);
+        currentDepth = addIteration(currentDepth, maxSubDagIteration);
 
         if (stepOperator instanceof StepLoopUntilOperator) {
             StepLoopUntilOperator loopUntilOperator = (StepLoopUntilOperator)stepOperator;

diff --git a/.../main/java/com/antgroup/geaflow/dsl/runtime/traversal/operator/StepLoopUntilOperator.java b/.../main/java/com/antgroup/geaflow/dsl/runtime/traversal/operator/StepLoopUntilOperator.java
@@ -17,14 +17,17 @@
 import com.antgroup.geaflow.dsl.common.data.RowEdge;
 import com.antgroup.geaflow.dsl.common.data.RowVertex;
 import com.antgroup.geaflow.dsl.common.data.StepRecord;
+import com.antgroup.geaflow.dsl.common.data.VirtualId;
 import com.antgroup.geaflow.dsl.runtime.function.graph.StepBoolFunction;
 import com.antgroup.geaflow.dsl.runtime.traversal.TraversalRuntimeContext;
 import com.antgroup.geaflow.dsl.runtime.traversal.collector.StepCollector;
 import com.antgroup.geaflow.dsl.runtime.traversal.collector.StepJumpCollector;
 import com.antgroup.geaflow.dsl.runtime.traversal.data.EndOfData;
+import com.antgroup.geaflow.dsl.runtime.traversal.data.IdOnlyVertex;
 import com.antgroup.geaflow.dsl.runtime.traversal.data.StepRecordWithPath;
 import com.antgroup.geaflow.dsl.runtime.traversal.data.VertexRecord;
 import com.antgroup.geaflow.dsl.runtime.traversal.path.ITreePath;
+import com.antgroup.geaflow.state.pushdown.filter.EmptyFilter;
 import java.util.List;
 
 public class StepLoopUntilOperator extends AbstractStepOperator<StepBoolFunction,
@@ -88,8 +91,17 @@ protected void processRecord(StepRecordWithPath record) {
     }
 
     private StepRecordWithPath selectLastLoopPath(StepRecordWithPath record, boolean fromZeroLoop) {
-        RowVertex vertexRecord = ((VertexRecord)record).getTreePath().getVertex();
+        RowVertex vertexId = ((VertexRecord)record).getVertex();
         if (fromZeroLoop) {
+            final RowVertex vertexRecord;
+            if (vertexId instanceof IdOnlyVertex && !(vertexId.getId() instanceof VirtualId)) {
+                vertexRecord = context.loadVertex(vertexId.getId(),
+                    EmptyFilter.of(),
+                    graphSchema,
+                    addingVertexFieldTypes);
+            } else {
+                vertexRecord = vertexId;
+            }
             return record.mapTreePath(treePath -> {
                 ITreePath newTreePath = treePath;
                 for (int i = 0; i < loopBodyPathFieldCount - 1; i++) {

diff --git a/...main/java/com/antgroup/geaflow/dsl/runtime/traversal/operator/StepNodeFilterOperator.java b/...main/java/com/antgroup/geaflow/dsl/runtime/traversal/operator/StepNodeFilterOperator.java
@@ -26,7 +26,7 @@ public StepNodeFilterOperator(long id, StepNodeFilterFunction function) {
 
     @Override
     protected void processRecord(VertexRecord record) {
-        if (function.filter(record.getTreePath().getVertex())) {
+        if (function.filter(record.getVertex())) {
             collect(record);
         }
     }

diff --git a/...ntime/src/main/java/com/antgroup/geaflow/dsl/runtime/traversal/path/AbstractTreePath.java b/...ntime/src/main/java/com/antgroup/geaflow/dsl/runtime/traversal/path/AbstractTreePath.java
@@ -197,15 +197,14 @@ protected ITreePath filter(PathFilterFunction filterFunction,
                                 }
                                 filterTrees.add(edgeTreePath);
                             }
-                        } else if (parentSize == 1) {
-                            ITreePath filterTree = ((AbstractTreePath) getParents().get(0)).filter(filterFunction,
-                                refPathIndices, fieldMapping, currentPath, maxDepth, pathId);
-                            if (!filterTree.isEmpty()) {
-                                filterTrees.add(filterTree.extendTo(edge));
+                        } else if (parentSize >= 1) {
+                            for (ITreePath parent : getParents()) {
+                                ITreePath filterTree = ((AbstractTreePath) parent).filter(filterFunction,
+                                    refPathIndices, fieldMapping, currentPath, maxDepth, pathId);
+                                if (!filterTree.isEmpty()) {
+                                    filterTrees.add(filterTree.extendTo(edge));
+                                }
                             }
-                        } else {
-                            throw new IllegalArgumentException("Edge node should have only one parent,"
-                                + " current is: " + parentSize);
                         }
                         currentPath.remove(currentPath.size() - 1);
                     }