[ISSUE-408] Support Incremental match operator (#470)

Author: 652053395

geaflow/geaflow-common/src/main/java/com/antgroup/geaflow/common/config/keys/DSLConfigKeys.java

@@ -20,6 +20,28 @@
 
 public class DSLConfigKeys implements Serializable {
 
+    private static final long serialVersionUID = 3550044668482560581L;
+
+    public static final ConfigKey INCR_TRAVERSAL_ITERATION_THRESHOLD = ConfigKeys
+        .key("geaflow.dsl.incr.traversal.iteration.threshold")
+        .defaultValue(4)
+        .description("The max iteration to enable incr match");
+
+    public static final ConfigKey TABLE_SINK_SPLIT_LINE = ConfigKeys
+        .key("geaflow.dsl.table.sink.split.line")
+        .noDefaultValue()
+        .description("The file sink split line.");
+
+    public static final ConfigKey ENABLE_INCR_TRAVERSAL = ConfigKeys
+        .key("geaflow.dsl.graph.enable.incr.traversal")
+        .defaultValue(false)
+        .description("Enable incr match");
+
+    public static final ConfigKey INCR_TRAVERSAL_WINDOW = ConfigKeys
+        .key("geaflow.dsl.graph.incr.traversal.window")
+        .defaultValue(-1L)
+        .description("When window id is large than this parameter to enable incr match");
+
     public static final ConfigKey GEAFLOW_DSL_STORE_TYPE = ConfigKeys
         .key("geaflow.dsl.graph.store.type")
         .noDefaultValue()

geaflow/geaflow-core/geaflow-runtime/geaflow-operator/src/main/java/com/antgroup/geaflow/operator/impl/graph/traversal/dynamic/AbstractDynamicGraphVertexCentricTraversalOp.java

@@ -176,12 +176,13 @@ public void close() {
         this.responses.clear();
     }
 
-    class IncGraphVCTraversalCtxImpl extends IncGraphContextImpl<K, VV, EV, M>
+    public class IncGraphVCTraversalCtxImpl extends IncGraphContextImpl<K, VV, EV, M>
         implements IncVertexCentricTraversalFuncContext<K, VV, EV, M, R> {
 
         private final ICollector<IGraphMessage<K, M>> messageCollector;
         private final String opName;
         private final TraversalHistoricalGraph<K, VV, EV> traversalHistoricalGraph;
+        private boolean enableIncrMatch;
 
         protected IncGraphVCTraversalCtxImpl(String opName,
                                              ICollector<IGraphMessage<K, M>> messageCollector) {
@@ -192,6 +193,15 @@ protected IncGraphVCTraversalCtxImpl(String opName,
                 (IncHistoricalGraph<K, VV, EV>) super.getHistoricalGraph());
         }
 
+        public boolean isEnableIncrMatch() {
+            return enableIncrMatch;
+        }
+
+        public IncGraphVCTraversalCtxImpl setEnableIncrMatch(boolean enableIncrMatch) {
+            this.enableIncrMatch = enableIncrMatch;
+            return this;
+        }
+
         @Override
         public void activeRequest(ITraversalRequest<K> request) {
 

geaflow/geaflow-core/geaflow-runtime/geaflow-operator/src/main/java/com/antgroup/geaflow/operator/impl/graph/traversal/dynamic/DynamicGraphHelper.java

@@ -0,0 +1,51 @@
+/*
+ * Copyright 2023 AntGroup CO., Ltd.
+ *
+ * Licensed 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.
+ */
+
+package com.antgroup.geaflow.operator.impl.graph.traversal.dynamic;
+
+
+import com.antgroup.geaflow.api.context.RuntimeContext;
+import com.antgroup.geaflow.common.config.Configuration;
+import com.antgroup.geaflow.common.config.keys.DSLConfigKeys;
+
+public class DynamicGraphHelper {
+
+    public static boolean enableIncrTraversal(int maxIterationCount, int startIdSize, Configuration configuration) {
+        if (configuration != null) {
+            boolean res = configuration.getBoolean(DSLConfigKeys.ENABLE_INCR_TRAVERSAL);
+            if (!res) {
+                return false;
+            }
+        }
+
+        int traversalThreshold = configuration.getInteger(DSLConfigKeys.INCR_TRAVERSAL_ITERATION_THRESHOLD);
+        // when maxIterationCount <=2 no need to include subGraph, since 1 hop is already included in the incr edges.
+        return maxIterationCount > 2 && maxIterationCount <= traversalThreshold && startIdSize == 0;
+    }
+
+    public static boolean enableIncrTraversalRuntime(RuntimeContext runtimeContext) {
+        long windowId = runtimeContext.getWindowId();
+        if (windowId == 1) {
+            // the first window not need evolve
+            return false;
+        }
+        long window = runtimeContext.getConfiguration().getLong(DSLConfigKeys.INCR_TRAVERSAL_WINDOW);
+        if (window == -1) {
+            // default do incr
+            return true;
+        } else {
+            return windowId > window;
+        }
+    }
+}

geaflow/geaflow-core/geaflow-runtime/geaflow-operator/src/main/java/com/antgroup/geaflow/operator/impl/graph/traversal/dynamic/DynamicGraphVertexCentricTraversalAllOp.java

@@ -33,16 +33,36 @@ public DynamicGraphVertexCentricTraversalAllOp(
         super(graphViewDesc, vcTraversal);
     }
 
+    private void traversalEvolveVIds() {
+        for (K vertexId : temporaryGraphCache.getAllEvolveVId()) {
+            ITraversalRequest<K> traversalRequest = new VertexBeginTraversalRequest<>(vertexId);
+            this.graphVCTraversalCtx.init(iterations, vertexId);
+            this.incVcTraversalFunction.init(traversalRequest);
+        }
+    }
+
     @Override
     protected void traversalByRequest() {
-        if (!temporaryGraphCache.getAllEvolveVId().isEmpty()) {
-            try (CloseableIterator<K> idIterator =
-                     graphState.dynamicGraph().V().query(GRAPH_VERSION, keyGroup).idIterator()) {
-                while (idIterator.hasNext()) {
-                    K vertexId = idIterator.next();
-                    ITraversalRequest<K> traversalRequest = new VertexBeginTraversalRequest<>(vertexId);
-                    this.graphVCTraversalCtx.init(iterations, vertexId);
-                    this.incVcTraversalFunction.init(traversalRequest);
+        if (graphVCTraversalCtx.isEnableIncrMatch() && DynamicGraphHelper.enableIncrTraversalRuntime(runtimeContext)) {
+            traversalEvolveVIds();
+
+        } else {
+            if (function.getMaxIterationCount() <= 2) {
+                // The evolved vertices/edges can cover the match pattern when iteration <= 2 (e.g match(a)->(b)).
+                traversalEvolveVIds();
+                return;
+            }
+
+            // Traversal all vertices.
+            if (!temporaryGraphCache.getAllEvolveVId().isEmpty()) {
+                try (CloseableIterator<K> idIterator =
+                    graphState.dynamicGraph().V().query(GRAPH_VERSION, keyGroup).idIterator()) {
+                    while (idIterator.hasNext()) {
+                        K vertexId = idIterator.next();
+                        ITraversalRequest<K> traversalRequest = new VertexBeginTraversalRequest<>(vertexId);
+                        this.graphVCTraversalCtx.init(iterations, vertexId);
+                        this.incVcTraversalFunction.init(traversalRequest);
+                    }
                 }
             }
         }

geaflow/geaflow-dsl/geaflow-dsl-connector/geaflow-dsl-connector-file/src/main/java/com/antgroup/geaflow/dsl/connector/file/sink/FileTableSink.java

@@ -17,6 +17,7 @@
 import com.antgroup.geaflow.api.context.RuntimeContext;
 import com.antgroup.geaflow.common.config.Configuration;
 import com.antgroup.geaflow.common.config.keys.ConnectorConfigKeys;
+import com.antgroup.geaflow.common.config.keys.DSLConfigKeys;
 import com.antgroup.geaflow.dsl.common.data.Row;
 import com.antgroup.geaflow.dsl.common.types.StructType;
 import com.antgroup.geaflow.dsl.connector.api.TableSink;
@@ -72,6 +73,11 @@ public void write(Row row) throws IOException {
 
     @Override
     public void finish() throws IOException {
+        String split = tableConf.getString(DSLConfigKeys.TABLE_SINK_SPLIT_LINE.getKey(), null);
+        if (split != null) {
+            writer.write(split + "\n");
+        }
+
         writer.flush();
     }
 

geaflow/geaflow-dsl/geaflow-dsl-runtime/src/main/java/com/antgroup/geaflow/dsl/runtime/engine/GeaFlowCommonTraversalFunction.java

@@ -123,4 +123,16 @@ public void finish(long iterationId) {
     public void close() {
         executeDagGroup.close();
     }
+
+    public ExecuteDagGroup getExecuteDagGroup() {
+        return executeDagGroup;
+    }
+
+    public TraversalRuntimeContext getContext() {
+        return context;
+    }
+
+    public List<ITraversalRequest<Object>> getInitRequests() {
+        return initRequests;
+    }
 }

geaflow/geaflow-dsl/geaflow-dsl-runtime/src/main/java/com/antgroup/geaflow/dsl/runtime/engine/GeaFlowDynamicVCTraversal.java

@@ -28,12 +28,16 @@ public class GeaFlowDynamicVCTraversal extends IncVertexCentricTraversal<Object,
 
     private final boolean isTraversalAllWithRequest;
 
+    private final boolean enableIncrTraversal;
+
     public GeaFlowDynamicVCTraversal(ExecuteDagGroup executeDagGroup,
                                      int maxTraversal,
-                                     boolean isTraversalAllWithRequest) {
+                                     boolean isTraversalAllWithRequest,
+                                     boolean enableIncrTraversal) {
         super(maxTraversal);
         this.executeDagGroup = executeDagGroup;
         this.isTraversalAllWithRequest = isTraversalAllWithRequest;
+        this.enableIncrTraversal = enableIncrTraversal;
     }
 
     @Override
@@ -43,6 +47,6 @@ public VertexCentricCombineFunction<MessageBox> getCombineFunction() {
 
     @Override
     public IncVertexCentricTraversalFunction<Object, Row, Row, MessageBox, ITreePath> getIncTraversalFunction() {
-        return new GeaFlowDynamicVCTraversalFunction(executeDagGroup, isTraversalAllWithRequest);
+        return new GeaFlowDynamicVCTraversalFunction(executeDagGroup, isTraversalAllWithRequest, enableIncrTraversal);
     }
 }