Unload unreachable op after run (#916)

* Unload unreachable op after run

analytical_engine/core/object/object_manager.h

@@ -47,11 +47,9 @@ class ObjectManager {
   }
 
   bl::result<void> RemoveObject(const std::string& id) {
-    if (objects.find(id) == objects.end()) {
-      RETURN_GS_ERROR(vineyard::ErrorCode::kInvalidOperationError,
-                      "Object " + id + " does not exist");
+    if (objects.find(id) != objects.end()) {
+      objects.erase(id);
     }
-    objects.erase(id);
     return {};
   }
 

python/graphscope/client/session.py

@@ -22,7 +22,6 @@
 import atexit
 import base64
 import contextlib
-import copy
 import json
 import logging
 import os
@@ -31,8 +30,8 @@
 import sys
 import threading
 import time
+import uuid
 import warnings
-from queue import Empty as EmptyQueue
 
 try:
     from kubernetes import client as kube_client
@@ -51,13 +50,10 @@
 from graphscope.deploy.hosts.cluster import HostsClusterLauncher
 from graphscope.deploy.kubernetes.cluster import KubernetesClusterLauncher
 from graphscope.framework.dag import Dag
-from graphscope.framework.errors import ConnectionError
 from graphscope.framework.errors import FatalError
-from graphscope.framework.errors import GRPCError
 from graphscope.framework.errors import InteractiveEngineInternalError
 from graphscope.framework.errors import InvalidArgumentError
 from graphscope.framework.errors import K8sError
-from graphscope.framework.errors import check_argument
 from graphscope.framework.graph import Graph
 from graphscope.framework.graph import GraphDAGNode
 from graphscope.framework.operation import Operation
@@ -187,7 +183,7 @@ def _rebuild_gremlin_results(
         result_set_dag_node = self._fetches[seq]
         return ResultSet(result_set_dag_node)
 
-    def wrapper_results(self, response: message_pb2.RunStepResponse):
+    def wrap_results(self, response: message_pb2.RunStepResponse):
         rets = list()
         for seq, op in enumerate(self._ops):
             for op_result in response.results:
@@ -235,6 +231,26 @@ def wrapper_results(self, response: message_pb2.RunStepResponse):
                     break
         return rets[0] if self._unpack else rets
 
+    def get_dag_for_unload(self):
+        """Unload operations (graph, app, context) in dag which are not
+        existed in fetches.
+        """
+        unload_dag = op_def_pb2.DagDef()
+        keys_of_fetches = set([op.key for op in self._ops])
+        mapping = {
+            types_pb2.CREATE_GRAPH: types_pb2.UNLOAD_GRAPH,
+            types_pb2.CREATE_APP: types_pb2.UNLOAD_APP,
+            types_pb2.RUN_APP: types_pb2.UNLOAD_CONTEXT,
+        }
+        for op_def in self._sub_dag.op:
+            if op_def.op in mapping and op_def.key not in keys_of_fetches:
+                unload_op_def = op_def_pb2.OpDef(
+                    op=mapping[op_def.op], key=uuid.uuid4().hex
+                )
+                unload_op_def.parents.extend([op_def.key])
+                unload_dag.op.extend([unload_op_def])
+        return unload_dag
+
 
 class Session(object):
     """A class for interacting with GraphScope graph computation service cluster.
@@ -936,7 +952,15 @@ def run(self, fetches, debug=False):
         except FatalError:
             self.close()
             raise
-        return fetch_handler.wrapper_results(response)
+        if not self.eager():
+            # Unload operations that cannot be touched anymore
+            dag_to_unload = fetch_handler.get_dag_for_unload()
+            try:
+                self._grpc_client.run(dag_to_unload)
+            except FatalError:
+                self.close()
+                raise
+        return fetch_handler.wrap_results(response)
 
     def _connect(self):
         if self._config_params["addr"] is not None:

python/graphscope/framework/dag_utils.py

@@ -677,6 +677,17 @@ def create_subgraph(graph, nodes=None, edges=None):
     return op
 
 
+def create_unload_op(session_id, op_type, inputs):
+    """Uility method to create a unload `Operation` based on op type and op."""
+    op = Operation(
+        session_id,
+        op_type,
+        inputs=inputs,
+        output_types=types_pb2.NULL_OUTPUT,
+    )
+    return op
+
+
 def unload_app(app):
     """Unload a loaded app.
 
@@ -686,13 +697,7 @@ def unload_app(app):
     Returns:
         An op to unload the `app`.
     """
-    op = Operation(
-        app.session_id,
-        types_pb2.UNLOAD_APP,
-        inputs=[app.op],
-        output_types=types_pb2.NULL_OUTPUT,
-    )
-    return op
+    return create_unload_op(app.session_id, types_pb2.UNLOAD_APP, [app.op])
 
 
 def unload_graph(graph):
@@ -704,23 +709,11 @@ def unload_graph(graph):
     Returns:
         An op to unload the `graph`.
     """
-    op = Operation(
-        graph.session_id,
-        types_pb2.UNLOAD_GRAPH,
-        inputs=[graph.op],
-        output_types=types_pb2.NULL_OUTPUT,
-    )
-    return op
+    return create_unload_op(graph.session_id, types_pb2.UNLOAD_GRAPH, [graph.op])
 
 
 def unload_context(context):
-    op = Operation(
-        context.session_id,
-        types_pb2.UNLOAD_CONTEXT,
-        inputs=[context.op],
-        output_types=types_pb2.NULL_OUTPUT,
-    )
-    return op
+    return create_unload_op(context.session_id, types_pb2.UNLOAD_CONTEXT, [context.op])
 
 
 def context_to_numpy(context, selector=None, vertex_range=None, axis=0):