[Runtime] Fix checkpoint causing empty results in PipelineServiceExecutorContext (#543) (#544)

* fix checkpoint causing empty results in interactive query
* skip checkpoint when context is PipelineServiceExecutorContext

Author: huhao0926

geaflow/geaflow-core/geaflow-runtime/geaflow-pipeline/src/main/java/com/antgroup/geaflow/runtime/pipeline/runner/PipelineRunner.java

@@ -106,11 +106,13 @@ private ICycleSchedulerContext loadOrCreateContext(IPipelineExecutorContext pipe
 
             Map<Integer, List<ExecutionTask>> vertex2Tasks = ExecutionCycleTaskAssigner.assign(graph);
 
+            // Skip checkpoint if it's a PipelineServiceExecutorContext
+            boolean skipCheckpoint = pipelineExecutorContext instanceof PipelineServiceExecutorContext;
             IExecutionCycle cycle = ExecutionCycleBuilder.buildExecutionCycle(graph, vertex2Tasks,
                 pipelineContext.getConfig(), ExecutionIdGenerator.getInstance().generateId(),
                 pipelineExecutorContext.getPipelineTaskId(), pipelineExecutorContext.getPipelineTaskName(),
                 ExecutionIdGenerator.getInstance().generateId(), pipelineExecutorContext.getDriverId(),
-                pipelineExecutorContext.getDriverIndex());
+                pipelineExecutorContext.getDriverIndex(), skipCheckpoint);
             return CycleSchedulerContextFactory.create(cycle, null);
         });
         return context;

geaflow/geaflow-core/geaflow-runtime/geaflow-runtime-core/src/main/java/com/antgroup/geaflow/runtime/core/scheduler/cycle/ExecutionCycleBuilder.java

@@ -45,7 +45,8 @@ public static IExecutionCycle buildExecutionCycle(ExecutionGraph executionGraph,
                                                       String name,
                                                       long schedulerId,
                                                       String driverId,
-                                                      int driverIndex) {
+                                                      int driverIndex,
+                                                      boolean skipCheckpoint) {
 
         int flyingCount = executionGraph.getCycleGroupMeta().getFlyingCount();
         long iterationCount = executionGraph.getCycleGroupMeta().getIterationCount();
@@ -54,7 +55,7 @@ public static IExecutionCycle buildExecutionCycle(ExecutionGraph executionGraph,
         for (ExecutionVertexGroup vertexGroup : executionGraph.getVertexGroupMap().values()) {
             ExecutionNodeCycle nodeCycle = buildExecutionCycle(vertexGroup,
                 vertex2Tasks, config, pipelineId, pipelineTaskId, name, schedulerId, driverId, driverIndex);
-            graphCycle.addCycle(nodeCycle);
+            graphCycle.addCycle(nodeCycle, skipCheckpoint);
         }
         return graphCycle;
     }

geaflow/geaflow-core/geaflow-runtime/geaflow-runtime-core/src/main/java/com/antgroup/geaflow/runtime/core/scheduler/cycle/ExecutionGraphCycle.java

@@ -68,7 +68,7 @@ public HighAvailableLevel getHighAvailableLevel() {
         return haLevel;
     }
 
-    public void addCycle(IExecutionCycle cycle) {
+    public void addCycle(IExecutionCycle cycle, boolean skipCheckpoint) {
         if (cycleMap.containsKey(cycle.getCycleId())) {
             throw new GeaflowRuntimeException(String.format("cycle %d already added", cycle.getCycleId()));
         }
@@ -80,8 +80,8 @@ public void addCycle(IExecutionCycle cycle) {
         cycleParents.get(cycle.getCycleId()).addAll(nodeCycle.getVertexGroup().getParentVertexGroupIds());
         cycleChildren.get(cycle.getCycleId()).addAll(nodeCycle.getVertexGroup().getChildrenVertexGroupIds());
 
-        if (iterationCount > 1 && haLevel != HighAvailableLevel.CHECKPOINT
-            && (cycle.getType() == ExecutionCycleType.ITERATION || cycle.getType() == ExecutionCycleType.ITERATION_WITH_AGG)) {
+        if (!skipCheckpoint && (iterationCount > 1 && haLevel != HighAvailableLevel.CHECKPOINT
+            && (cycle.getType() == ExecutionCycleType.ITERATION || cycle.getType() == ExecutionCycleType.ITERATION_WITH_AGG))) {
             haLevel = HighAvailableLevel.CHECKPOINT;
         }
     }

geaflow/geaflow-core/geaflow-runtime/geaflow-runtime-core/src/test/java/com/antgroup/geaflow/runtime/core/scheduler/ExecutionGraphCycleSchedulerTest.java

@@ -206,7 +206,7 @@ private ExecutionGraphCycle buildMockGraphCycle() {
             1, 10, configuration, "driver_id", 0);
         ExecutionNodeCycle nodeCycle = buildMockIterationNodeCycle(configuration);
 
-        graphCycle.addCycle(nodeCycle);
+        graphCycle.addCycle(nodeCycle, false);
         try {
             ReflectionUtil.setField(graphCycle, "haLevel", CHECKPOINT);
         } catch (Exception e) {

geaflow/geaflow-core/geaflow-runtime/geaflow-runtime-core/src/test/java/com/antgroup/geaflow/runtime/core/scheduler/context/ExecutionGraphCycleSchedulerContextTest.java

@@ -40,7 +40,7 @@ public void testExecutionGraphWithCheckpointChileCycle() {
         parentContext.init(1);
 
         ExecutionNodeCycle iterationCycle = buildPipelineCycle(false, finishIterationId);
-        graph.addCycle(iterationCycle);
+        graph.addCycle(iterationCycle, false);
         CheckpointSchedulerContext iterationContext = new CheckpointSchedulerContext(iterationCycle, parentContext);
         iterationContext.init();
 

geaflow/geaflow-core/geaflow-runtime/geaflow-runtime-core/src/test/java/com/antgroup/geaflow/runtime/core/scheduler/cycle/ExecutionCycleBuilderTest.java

@@ -35,7 +35,7 @@ public void validateGraphCycleFlyingCount() {
         meta.setFlyingCount(0);
         meta.setGroupType(CycleGroupType.windowed);
 
-        ExecutionCycleBuilder.buildExecutionCycle(executionGraph, null, null, 0, 0, null, 0,null, 0);
+        ExecutionCycleBuilder.buildExecutionCycle(executionGraph, null, null, 0, 0, null, 0,null, 0, false);
     }
 
     @Test(expectedExceptions = IllegalArgumentException.class,
@@ -45,7 +45,7 @@ public void validateGraphCycleIterationCount() {
         CycleGroupMeta meta = executionGraph.getCycleGroupMeta();
         meta.setIterationCount(0);
         meta.setGroupType(CycleGroupType.windowed);
-        ExecutionCycleBuilder.buildExecutionCycle(executionGraph, null, null, 0, 0, null, 0, null, 0);
+        ExecutionCycleBuilder.buildExecutionCycle(executionGraph, null, null, 0, 0, null, 0, null, 0, false);
     }
 
     @Test(expectedExceptions = IllegalArgumentException.class,
@@ -62,7 +62,7 @@ public void validateNodeCycleFlyingCount() {
         executionGraph.getVertexGroupMap().put(1, vertexGroup);
 
         ExecutionCycleBuilder.buildExecutionCycle(executionGraph, null, null, 0, 0,  null, 0, null,
-            0);
+            0, false);
     }
 
     @Test(expectedExceptions = IllegalArgumentException.class,
@@ -79,6 +79,6 @@ public void validateNodeCycleIterationCount() {
         executionGraph.getVertexGroupMap().put(1, vertexGroup);
 
         ExecutionCycleBuilder.buildExecutionCycle(executionGraph, null, null, 0, 0,  null, 0, null,
-            0);
+            0, false);
     }
 }