Support transformation of directed/undirected graph (#1101)

* Add transform graph
* fix bugs
* debug image
* update  vineyard version to 0.3.12
* revert debug image

.github/workflows/gae.yml

@@ -32,7 +32,7 @@ jobs:
   build-gae:
     runs-on: ubuntu-20.04
     container:
-      image: registry.cn-hongkong.aliyuncs.com/graphscope/graphscope-vineyard:v0.3.11
+      image: registry.cn-hongkong.aliyuncs.com/graphscope/graphscope-vineyard:v0.3.12
       options:
         --shm-size 4096m
     steps:

.github/workflows/gss.yml

@@ -39,7 +39,7 @@ jobs:
     # be configured manually when a new self-hosted runner is added.
     runs-on: self-hosted
     container:
-      image: registry.cn-hongkong.aliyuncs.com/graphscope/graphscope-vineyard:v0.3.11
+      image: registry.cn-hongkong.aliyuncs.com/graphscope/graphscope-vineyard:v0.3.12
     defaults:
       run:
         shell: bash --noprofile --norc -eo pipefail {0}

.github/workflows/networkx-forward-algo-nightly.yml

@@ -16,7 +16,7 @@ jobs:
       run:
         shell: bash --noprofile --norc -eo pipefail {0}
     container:
-      image: registry.cn-hongkong.aliyuncs.com/graphscope/graphscope-vineyard:v0.3.11
+      image: registry.cn-hongkong.aliyuncs.com/graphscope/graphscope-vineyard:v0.3.12
       options:
         --shm-size 4096m
 

analytical_engine/CMakeLists.txt

@@ -153,7 +153,7 @@ endif ()
 find_package(libgrapelite REQUIRED)
 include_directories(${LIBGRAPELITE_INCLUDE_DIRS})
 
-find_package(vineyard 0.3.11 REQUIRED)
+find_package(vineyard 0.3.12 REQUIRED)
 include_directories(${VINEYARD_INCLUDE_DIRS})
 add_compile_options(-DENABLE_SELECTOR)
 

analytical_engine/core/grape_instance.cc

@@ -776,7 +776,6 @@ bl::result<rpc::graph::GraphDefPb> GrapeInstance::copyGraph(
 
 bl::result<rpc::graph::GraphDefPb> GrapeInstance::toDirected(
     const rpc::GSParams& params) {
-#ifdef NETWORKX
   BOOST_LEAF_AUTO(src_graph_name, params.Get<std::string>(rpc::GRAPH_NAME));
   // BOOST_LEAF_AUTO(copy_type, params.Get<std::string>(rpc::COPY_TYPE));
 
@@ -788,16 +787,10 @@ bl::result<rpc::graph::GraphDefPb> GrapeInstance::toDirected(
                   src_wrapper->ToDirected(comm_spec_, dst_graph_name));
   BOOST_LEAF_CHECK(object_manager_.PutObject(dst_wrapper));
   return dst_wrapper->graph_def();
-#else
-  RETURN_GS_ERROR(vineyard::ErrorCode::kUnimplementedMethod,
-                  "GraphScope is built with NETWORKX=OFF, please recompile it "
-                  "with NETWORKX=ON");
-#endif  // NETWORKX
 }
 
 bl::result<rpc::graph::GraphDefPb> GrapeInstance::toUnDirected(
     const rpc::GSParams& params) {
-#ifdef NETWORKX
   BOOST_LEAF_AUTO(src_graph_name, params.Get<std::string>(rpc::GRAPH_NAME));
 
   BOOST_LEAF_AUTO(src_wrapper,
@@ -808,11 +801,6 @@ bl::result<rpc::graph::GraphDefPb> GrapeInstance::toUnDirected(
                   src_wrapper->ToUndirected(comm_spec_, dst_graph_name));
   BOOST_LEAF_CHECK(object_manager_.PutObject(dst_wrapper));
   return dst_wrapper->graph_def();
-#else
-  RETURN_GS_ERROR(vineyard::ErrorCode::kUnimplementedMethod,
-                  "GraphScope is built with NETWORKX=OFF, please recompile it "
-                  "with NETWORKX=ON");
-#endif  // NETWORKX
 }
 
 #ifdef NETWORKX

analytical_engine/core/object/fragment_wrapper.h

@@ -635,15 +635,40 @@ class FragmentWrapper<vineyard::ArrowFragment<OID_T, VID_T>>
   bl::result<std::shared_ptr<IFragmentWrapper>> ToDirected(
       const grape::CommSpec& comm_spec,
       const std::string& dst_graph_name) override {
-    RETURN_GS_ERROR(vineyard::ErrorCode::kInvalidOperationError,
-                    "Cannot convert to the directed ArrowFragment");
+    auto& meta = fragment_->meta();
+    auto* client = dynamic_cast<vineyard::Client*>(meta.GetClient());
+    int thread_num =
+        (std::thread::hardware_concurrency() + comm_spec.local_num() - 1) /
+        comm_spec.local_num();
+    BOOST_LEAF_AUTO(new_frag_id,
+                    fragment_->TransformDirection(*client, thread_num));
+    VINEYARD_CHECK_OK(client->Persist(new_frag_id));
+    BOOST_LEAF_AUTO(frag_group_id, vineyard::ConstructFragmentGroup(
+                                       *client, new_frag_id, comm_spec));
+    auto new_frag = client->GetObject<fragment_t>(new_frag_id);
+
+    rpc::graph::GraphDefPb new_graph_def;
+
+    new_graph_def.set_key(dst_graph_name);
+
+    gs::rpc::graph::VineyardInfoPb vy_info;
+    if (graph_def_.has_extension()) {
+      graph_def_.extension().UnpackTo(&vy_info);
+    }
+    vy_info.set_vineyard_id(frag_group_id);
+    new_graph_def.mutable_extension()->PackFrom(vy_info);
+
+    set_graph_def(new_frag, new_graph_def);
+
+    auto wrapper = std::make_shared<FragmentWrapper<fragment_t>>(
+        dst_graph_name, new_graph_def, new_frag);
+    return std::dynamic_pointer_cast<IFragmentWrapper>(wrapper);
   }
 
   bl::result<std::shared_ptr<IFragmentWrapper>> ToUndirected(
       const grape::CommSpec& comm_spec,
       const std::string& dst_graph_name) override {
-    RETURN_GS_ERROR(vineyard::ErrorCode::kInvalidOperationError,
-                    "Cannot convert to the undirected ArrowFragment");
+    return ToDirected(comm_spec, dst_graph_name);
   }
 
   bl::result<std::shared_ptr<IFragmentWrapper>> CreateGraphView(

coordinator/gscoordinator/template/CMakeLists.template

@@ -153,7 +153,7 @@ if (libgrapelite_FOUND)
 endif()
 
 # find vineyard-----------------------------------------
-find_package(vineyard 0.3.11 QUIET)
+find_package(vineyard 0.3.12 QUIET)
 if (vineyard_FOUND)
   include_directories(AFTER SYSTEM ${VINEYARD_INCLUDE_DIRS})
 endif()

coordinator/gscoordinator/utils.py

@@ -502,6 +502,8 @@ def op_pre_process(op, op_result_pool, key_to_op, **kwargs):  # noqa: C901
         )
     if op.op == types_pb2.OUTPUT:
         _pre_process_for_output_op(op, op_result_pool, key_to_op, **kwargs)
+    if op.op in (types_pb2.TO_DIRECTED, types_pb2.TO_UNDIRECTED):
+        _pre_process_for_transform_op(op, op_result_pool, key_to_op, **kwargs)
 
 
 def _pre_process_for_create_graph_op(op, op_result_pool, key_to_op, **kwargs):
@@ -528,6 +530,14 @@ def _pre_process_for_add_labels_op(op, op_result_pool, key_to_op, **kwargs):
             )
 
 
+def _pre_process_for_transform_op(op, op_result_pool, key_to_op, **kwargs):
+    assert len(op.parents) == 1
+    result = op_result_pool[op.parents[0]]
+    # To compatible with eager evaluation cases where it will has the key.
+    if types_pb2.GRAPH_NAME not in op.attr:
+        op.attr[types_pb2.GRAPH_NAME].CopyFrom(utils.s_to_attr(result.graph_def.key))
+
+
 def _pre_process_for_close_interactive_query_op(
     op, op_result_pool, key_to_op, **kwargs
 ):
@@ -733,6 +743,8 @@ def __backtrack_key_of_graph_op(key):
                     types_pb2.TRANSFORM_GRAPH,
                     types_pb2.PROJECT_GRAPH,
                     types_pb2.PROJECT_TO_SIMPLE,
+                    types_pb2.TO_DIRECTED,
+                    types_pb2.TO_UNDIRECTED,
                 ):
                     return next_op
                 for parent_key in next_op.parents:

k8s/Makefile

@@ -40,7 +40,7 @@ else
 endif
 
 VERSION ?= latest
-VINEYARD_VERSION ?= v0.3.11
+VINEYARD_VERSION ?= v0.3.12
 PROFILE ?= release
 
 

k8s/graphscope-dev.Dockerfile

@@ -4,7 +4,7 @@
 # the result image includes all runtime stuffs of graphscope, with analytical engine,
 # learning engine and interactive engine installed.
 
-ARG BASE_VERSION=v0.3.11
+ARG BASE_VERSION=v0.3.12
 FROM registry.cn-hongkong.aliyuncs.com/graphscope/graphscope-vineyard:$BASE_VERSION as builder
 
 ARG NETWORKX=ON

k8s/graphscope-store.Dockerfile

@@ -1,4 +1,4 @@
-ARG BASE_VERSION=v0.3.11
+ARG BASE_VERSION=v0.3.12
 FROM registry.cn-hongkong.aliyuncs.com/graphscope/graphscope-vineyard:$BASE_VERSION as builder
 
 ARG CI=true

k8s/gsvineyard.Dockerfile

@@ -16,7 +16,7 @@ RUN sudo mkdir -p /opt/vineyard && \
     make -j`nproc` && \
     make install && \
     cd /tmp && \
-    git clone -b v0.3.11 https://github.com/alibaba/libvineyard.git --depth=1 && \
+    git clone -b v0.3.12 https://github.com/alibaba/libvineyard.git --depth=1 && \
     cd libvineyard && \
     git submodule update --init && \
     mkdir -p /tmp/libvineyard/build && \

k8s/ubuntu/gsvineyard.Dockerfile

@@ -14,7 +14,7 @@ RUN cd /tmp && \
     make -j`nproc` && \
     make install && \
     cd /tmp && \
-    git clone -b v0.3.11 https://github.com/alibaba/libvineyard.git --depth=1 && \
+    git clone -b v0.3.12 https://github.com/alibaba/libvineyard.git --depth=1 && \
     cd libvineyard && \
     git submodule update --init && \
     mkdir -p /tmp/libvineyard/build && \

python/graphscope/framework/dag_utils.py

@@ -572,46 +572,56 @@ def copy_graph(graph, copy_type="identical"):
 
 
 def to_directed(graph):
-    """Create to_directed operation for nx graph.
+    """Create to_directed operation graph.
 
     Args:
-        graph (:class:`nx.Graph`): A nx graph.
+        graph (:class:`nx.Graph`)
 
     Returns:
         Operation
     """
-    check_argument(graph.graph_type == graph_def_pb2.DYNAMIC_PROPERTY)
-    config = {
-        types_pb2.GRAPH_NAME: utils.s_to_attr(graph.key),
-    }
+    check_argument(
+        graph.graph_type
+        in (graph_def_pb2.DYNAMIC_PROPERTY, graph_def_pb2.ARROW_PROPERTY)
+    )
+    config = {}
+    # The key maybe filled later in coordinator
+    if hasattr(graph, "key"):
+        config[types_pb2.GRAPH_NAME] = utils.s_to_attr(graph.key)
 
     op = Operation(
         graph.session_id,
         types_pb2.TO_DIRECTED,
         config=config,
+        inputs=[graph.op],
         output_types=types_pb2.GRAPH,
     )
     return op
 
 
 def to_undirected(graph):
-    """Create to_undirected operation for nx graph.
+    """Create to_undirected operation for graph.
 
     Args:
-        graph (:class:`nx.Graph`): A nx graph.
+        graph (:class:`nx.Graph`)
 
     Returns:
         Operation
     """
-    check_argument(graph.graph_type == graph_def_pb2.DYNAMIC_PROPERTY)
-    config = {
-        types_pb2.GRAPH_NAME: utils.s_to_attr(graph.key),
-    }
+    check_argument(
+        graph.graph_type
+        in (graph_def_pb2.DYNAMIC_PROPERTY, graph_def_pb2.ARROW_PROPERTY)
+    )
+    config = {}
+    # The key maybe filled later in coordinator
+    if hasattr(graph, "key"):
+        config[types_pb2.GRAPH_NAME] = utils.s_to_attr(graph.key)
 
     op = Operation(
         graph.session_id,
         types_pb2.TO_UNDIRECTED,
         config=config,
+        inputs=[graph.op],
         output_types=types_pb2.GRAPH,
     )
     return op

python/graphscope/framework/graph.py

@@ -367,6 +367,16 @@ def to_dataframe(self, selector, vertex_range=None):
         op = dag_utils.graph_to_dataframe(self, selector, vertex_range)
         return ResultDAGNode(self, op)
 
+    def to_directed(self):
+        op = dag_utils.to_directed(self)
+        graph_dag_node = GraphDAGNode(self._session, op)
+        return graph_dag_node
+
+    def to_undirected(self):
+        op = dag_utils.to_undirected(self)
+        graph_dag_node = GraphDAGNode(self._session, op)
+        return graph_dag_node
+
     def add_vertices(self, vertices, label="_", properties=None, vid_field=0):
         """Add vertices to the graph, and return a new graph.
 
@@ -903,6 +913,31 @@ def to_dataframe(self, selector, vertex_range=None):
             self._graph_node.to_dataframe(selector, vertex_range)
         )
 
+    def to_directed(self):
+        """Returns a directed representation of the graph.
+
+        Returns:
+            :class:`Graph`: A directed graph with the same name, same nodes, and
+                with each edge (u, v, data) replaced by two directed edges (u, v, data) and (v, u, data).
+
+        """
+        if self._directed:
+            return self
+        return self._session._wrapper(self._graph_node.to_directed())
+
+    def to_undirected(self):
+        """Returns an undirected representation of the digraph.
+
+        Returns:
+            :class:`Graph`: An undirected graph with the same name and nodes and
+                with edge (u, v, data) if either (u, v, data) or (v, u, data) is in the digraph.
+                If both edges exist in digraph, they will both be preserved.
+                You must check and correct for this manually if desired.
+        """
+        if not self._directed:
+            return self
+        return self._session._wrapper(self._graph_node.to_undirected())
+
     def is_directed(self):
         return self._directed
 

python/graphscope/nx/classes/graph.py

@@ -1851,6 +1851,7 @@ def subgraph(self, nodes):
         g._key = graph_def.key
         g._session = self._session
         g._schema = copy.deepcopy(self._schema)
+        g._op = op
         return g
 
     def edge_subgraph(self, edges):

python/graphscope/tests/unittest/test_graph.py

@@ -532,6 +532,14 @@ def test_error_on_project(arrow_property_graph, ldbc_graph):
         sub_graph._project_to_simple()
 
 
+def test_transform(arrow_modern_graph):
+    g = arrow_modern_graph.to_undirected()
+    assert not g.is_directed()
+
+    g2 = g.to_directed()
+    assert g2.is_directed()
+
+
 def test_add_column(ldbc_graph, arrow_modern_graph):
     ldbc = ldbc_graph
     modern = arrow_modern_graph

scripts/install_deps.sh

@@ -14,8 +14,8 @@ readonly GREEN="\033[0;32m"
 readonly NC="\033[0m" # No Color
 
 readonly GRAPE_BRANCH="master" # libgrape-lite branch
-readonly V6D_VERSION="0.3.11"  # vineyard version
-readonly V6D_BRANCH="v0.3.11" # vineyard branch
+readonly V6D_VERSION="0.3.12"  # vineyard version
+readonly V6D_BRANCH="v0.3.12" # vineyard branch
 
 readonly OUTPUT_ENV_FILE="${HOME}/.graphscope_env"
 IS_IN_WSL=false && [[ ! -z "${IS_WSL}" || ! -z "${WSL_DISTRO_NAME}" ]] && IS_IN_WSL=true