Cugraph-Service Remote Graphs and Algorithm Dispatch

python/cugraph/cugraph/gnn/pyg_extensions/loader/dispatch.py

@@ -11,23 +11,76 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-from cugraph.structure.graph_implementation import (
-    simpleDistributedGraphImpl,
-    simpleGraphImpl,
-)
+# cuGraph or cuGraph-Service is required; each has its own version of
+# import_optional and we need to select the correct one.
+try:
+    from cugraph_service.client.remote_graph_utils import import_optional
+except ModuleNotFoundError:
+    try:
+        from cugraph.utilities.utils import import_optional
+    except ModuleNotFoundError:
+        raise ModuleNotFoundError(
+            "cuGraph extensions for PyG require cuGraph"
+            "or cuGraph-Service to be installed."
+        )
 
+_transform_to_backend_dtype_1d = import_optional("_transform_to_backend_dtype_1d")
+cudf = import_optional("cudf")
+pandas = import_optional("pandas")
+
+
+def call_cugraph_algorithm(name, graph, *args, backend="numpy", **kwargs):
+    """
+    Calls a cugraph algorithm for a remote, sg, or mg graph.
+    Requires either cuGraph or cuGraph-Service to be installed.
+
+    name : string
+        The name of the cuGraph algorithm to run (i.e. uniform_neighbor_sample)
+    graph : Graph (cuGraph) or RemoteGraph (cuGraph-Service)
+        The graph to call the algorithm on.
+    backend : ('cudf', 'pandas', 'cupy', 'numpy', 'torch', 'torch:<device>')
+              [default = 'numpy']
+        The backend where the algorithm results will be stored.  Only used
+        if the graph is a remote graph.
+    """
+
+    if graph.is_remote():
+        # If the graph is remote, cuGraph-Service must be installed
+        # Therefore we do not explicitly check that it is available
+        if name != "uniform_neighbor_sample":
+            raise ValueError(
+                f"cuGraph algorithm {name} is not yet supported for RemoteGraph"
+            )
+        else:
+            # TODO eventually replace this with a "call_algorithm call"
+            sample_result = graph._client.uniform_neighbor_sample(
+                *args, **kwargs, graph_id=graph._graph_id
+            )
+
+            if backend == "cudf":
+                df = cudf.DataFrame()
+            elif backend == "pandas":
+                df = pandas.DataFrame()
+            else:
+                # handle cupy, numpy, torch as dict of arrays/tensors
+                df = {}
+
+            # _transform_to_backend_dtype_1d handles array/Series conversion
+            for k, v in sample_result.__dict__.items():
+                df[k] = _transform_to_backend_dtype_1d(
+                    v, series_name=k, backend=backend
+                )
+
+            return df
 
-def call_cugraph_algorithm(name, graph, *args, **kwargs):
     # TODO check using graph property in a future PR
-    if isinstance(graph._Impl, simpleDistributedGraphImpl):
+    elif graph.is_multi_gpu():
         import cugraph.dask
 
         return getattr(cugraph.dask, name)(graph, *args, **kwargs)
 
     # TODO check using graph property in a future PR
-    elif isinstance(graph._Impl, simpleGraphImpl):
+    else:
         import cugraph
 
         return getattr(cugraph, name)(graph, *args, **kwargs)
-
-    # TODO Properly dispatch for cugraph-service.

python/cugraph/cugraph/structure/graph_classes.py

@@ -593,6 +593,13 @@ def is_remote(self):
         """
         return False
 
+    def is_multi_gpu(self):
+        """
+        Returns True if the graph is a multi-gpu graph; otherwise
+        returns False.
+        """
+        return isinstance(self._Impl, simpleDistributedGraphImpl)
+
     def to_directed(self):
         """
         Return a directed representation of the graph.

python/cugraph/cugraph/structure/graph_implementation/simpleGraph.py

@@ -36,12 +36,14 @@ class simpleGraphImpl:
     edgeWeightCol = "weights"
     edgeIdCol = "edge_id"
     edgeTypeCol = "edge_type"
+    srcCol = "src"
+    dstCol = "dst"
 
     class EdgeList:
         def __init__(self, source, destination, edge_attr=None):
             self.edgelist_df = cudf.DataFrame()
-            self.edgelist_df["src"] = source
-            self.edgelist_df["dst"] = destination
+            self.edgelist_df[simpleGraphImpl.srcCol] = source
+            self.edgelist_df[simpleGraphImpl.dstCol] = destination
             self.weights = False
             if edge_attr is not None:
                 self.weights = True
@@ -245,7 +247,12 @@ def __from_edgelist(
             value_col=value_col, store_transposed=store_transposed, renumber=renumber
         )
 
-    def to_pandas_edgelist(self, source="src", destination="dst", weight="weights"):
+    def to_pandas_edgelist(
+        self,
+        source="src",
+        destination="dst",
+        weight="weights",
+    ):
         """
         Returns the graph edge list as a Pandas DataFrame.
 
@@ -266,11 +273,21 @@ def to_pandas_edgelist(self, source="src", destination="dst", weight="weights"):
         gdf = self.view_edge_list()
         if self.properties.weighted:
             gdf.rename(
-                columns={"src": source, "dst": destination, "weight": weight},
+                columns={
+                    simpleGraphImpl.srcCol: source,
+                    simpleGraphImpl.dstCol: destination,
+                    "weight": weight,
+                },
                 inplace=True,
             )
         else:
-            gdf.rename(columns={"src": source, "dst": destination}, inplace=True)
+            gdf.rename(
+                columns={
+                    simpleGraphImpl.srcCol: source,
+                    simpleGraphImpl.dstCol: destination,
+                },
+                inplace=True,
+            )
         return gdf.to_pandas()
 
     def to_pandas_adjacency(self):
@@ -296,9 +313,9 @@ def to_numpy_array(self):
         df = self.edgelist.edgelist_df
         np_array = np.full((nlen, nlen), 0.0)
         for i in range(0, elen):
-            np_array[df["src"].iloc[i], df["dst"].iloc[i]] = df[
-                self.edgeWeightCol
-            ].iloc[i]
+            np_array[
+                df[simpleGraphImpl.srcCol].iloc[i], df[simpleGraphImpl.dstCol].iloc[i]
+            ] = df[self.edgeWeightCol].iloc[i]
         return np_array
 
     def to_numpy_matrix(self):
@@ -345,11 +362,18 @@ def view_edge_list(self):
         edgelist_df = self.edgelist.edgelist_df
 
         if self.properties.renumbered:
-            edgelist_df = self.renumber_map.unrenumber(edgelist_df, "src")
-            edgelist_df = self.renumber_map.unrenumber(edgelist_df, "dst")
+            edgelist_df = self.renumber_map.unrenumber(
+                edgelist_df, simpleGraphImpl.srcCol
+            )
+            edgelist_df = self.renumber_map.unrenumber(
+                edgelist_df, simpleGraphImpl.dstCol
+            )
 
         if not self.properties.directed:
-            edgelist_df = edgelist_df[edgelist_df["src"] <= edgelist_df["dst"]]
+            edgelist_df = edgelist_df[
+                edgelist_df[simpleGraphImpl.srcCol]
+                <= edgelist_df[simpleGraphImpl.dstCol]
+            ]
             edgelist_df = edgelist_df.reset_index(drop=True)
             self.properties.edge_count = len(edgelist_df)
 
@@ -576,7 +600,9 @@ def number_of_vertices(self):
             elif self.transposedadjlist is not None:
                 self.properties.node_count = len(self.transposedadjlist.offsets) - 1
             elif self.edgelist is not None:
-                df = self.edgelist.edgelist_df[["src", "dst"]]
+                df = self.edgelist.edgelist_df[
+                    [simpleGraphImpl.srcCol, simpleGraphImpl.dstCol]
+                ]
                 self.properties.node_count = df.max().max() + 1
             else:
                 raise RuntimeError("Graph is Empty")
@@ -601,8 +627,8 @@ def number_of_edges(self, directed_edges=False):
                 if self.properties.directed is False:
                     self.properties.edge_count = len(
                         self.edgelist.edgelist_df[
-                            self.edgelist.edgelist_df["src"]
-                            >= self.edgelist.edgelist_df["dst"]
+                            self.edgelist.edgelist_df[simpleGraphImpl.srcCol]
+                            >= self.edgelist.edgelist_df[simpleGraphImpl.dstCol]
                         ]
                     )
                 else:
@@ -852,8 +878,8 @@ def _make_plc_graph(self, value_col=None, store_transposed=False, renumber=True)
         self._plc_graph = SGGraph(
             resource_handle=ResourceHandle(),
             graph_properties=graph_props,
-            src_array=self.edgelist.edgelist_df["src"],
-            dst_array=self.edgelist.edgelist_df["dst"],
+            src_array=self.edgelist.edgelist_df[simpleGraphImpl.srcCol],
+            dst_array=self.edgelist.edgelist_df[simpleGraphImpl.dstCol],
             weight_array=weight_col,
             edge_id_array=id_col,
             edge_type_array=type_col,
@@ -901,10 +927,15 @@ def to_undirected(self, G, store_transposed=False):
             df = self.edgelist.edgelist_df
             if self.edgelist.weights:
                 source_col, dest_col, value_col = symmetrize(
-                    df, "src", "dst", simpleGraphImpl.edgeWeightCol
+                    df,
+                    simpleGraphImpl.srcCol,
+                    simpleGraphImpl.dstCol,
+                    simpleGraphImpl.edgeWeightCol,
                 )
             else:
-                source_col, dest_col = symmetrize(df, "src", "dst")
+                source_col, dest_col = symmetrize(
+                    df, simpleGraphImpl.srcCol, simpleGraphImpl.dstCol
+                )
                 value_col = None
             G.edgelist = simpleGraphImpl.EdgeList(source_col, dest_col, value_col)
 
@@ -923,25 +954,29 @@ def has_node(self, n):
             tmp = self.renumber_map.to_internal_vertex_id(cudf.Series([n]))
             return tmp[0] is not cudf.NA and tmp[0] >= 0
         else:
-            df = self.edgelist.edgelist_df[["src", "dst"]]
+            df = self.edgelist.edgelist_df[
+                [simpleGraphImpl.srcCol, simpleGraphImpl.dstCol]
+            ]
             return (df == n).any().any()
 
     def has_edge(self, u, v):
         """
         Returns True if the graph contains the edge (u,v).
         """
         if self.properties.renumbered:
-            tmp = cudf.DataFrame({"src": [u, v]})
-            tmp = tmp.astype({"src": "int"})
+            tmp = cudf.DataFrame({simpleGraphImpl.srcCol: [u, v]})
+            tmp = tmp.astype({simpleGraphImpl.srcCol: "int"})
             tmp = self.renumber_map.add_internal_vertex_id(
-                tmp, "id", "src", preserve_order=True
+                tmp, "id", simpleGraphImpl.srcCol, preserve_order=True
             )
 
             u = tmp["id"][0]
             v = tmp["id"][1]
 
         df = self.edgelist.edgelist_df
-        return ((df["src"] == u) & (df["dst"] == v)).any()
+        return (
+            (df[simpleGraphImpl.srcCol] == u) & (df[simpleGraphImpl.dstCol] == v)
+        ).any()
 
     def has_self_loop(self):
         """
@@ -950,7 +985,7 @@ def has_self_loop(self):
         # Detect self loop
         if self.properties.self_loop is None:
             elist = self.edgelist.edgelist_df
-            if (elist["src"] == elist["dst"]).any():
+            if (elist[simpleGraphImpl.srcCol] == elist[simpleGraphImpl.dstCol]).any():
                 self.properties.self_loop = True
             else:
                 self.properties.self_loop = False
@@ -962,7 +997,7 @@ def edges(self):
         sources and destinations. It does not return the edge weights.
         For viewing edges with weights use view_edge_list()
         """
-        return self.view_edge_list()[["src", "dst"]]
+        return self.view_edge_list()[[simpleGraphImpl.srcCol, simpleGraphImpl.dstCol]]
 
     def nodes(self):
         """
@@ -981,7 +1016,9 @@ def nodes(self):
                 else:
                     return df[df.columns[0]]
             else:
-                return cudf.concat([df["src"], df["dst"]]).unique()
+                return cudf.concat(
+                    [df[simpleGraphImpl.srcCol], df[simpleGraphImpl.dstCol]]
+                ).unique()
         if self.adjlist is not None:
             return cudf.Series(np.arange(0, self.number_of_nodes()))
 
@@ -995,7 +1032,9 @@ def neighbors(self, n):
             n = node[0]
 
         df = self.edgelist.edgelist_df
-        neighbors = df[df["src"] == n]["dst"].reset_index(drop=True)
+        neighbors = df[df[simpleGraphImpl.srcCol] == n][
+            simpleGraphImpl.dstCol
+        ].reset_index(drop=True)
         if self.properties.renumbered:
             # FIXME:  Multi-column vertices
             return self.renumber_map.from_internal_vertex_id(neighbors)["0"]

python/cugraph_service/cugraph_service_client/__init__.py

@@ -34,4 +34,3 @@
 
 from cugraph_service_client.client import CugraphServiceClient
 from cugraph_service_client.remote_graph import RemoteGraph
-from cugraph_service_client.remote_graph import RemotePropertyGraph

python/cugraph_service/cugraph_service_client/client.py

@@ -22,8 +22,8 @@
 import cupy as cp
 
 from cugraph_service_client import defaults
+from cugraph_service_client.remote_graph import RemoteGraph
 from cugraph_service_client import extension_return_dtype_map
-from cugraph_service_client.remote_graph import RemotePropertyGraph
 from cugraph_service_client.types import (
     ValueWrapper,
     GraphVertexEdgeID,
@@ -515,9 +515,9 @@ def delete_graph(self, graph_id):
 
     def graph(self):
         """
-        Constructs an empty RemotePropertyGraph object.
+        Constructs a new RemoteGraph object wrapping a remote PropertyGraph.
         """
-        return RemotePropertyGraph(self, self.create_graph())
+        return RemoteGraph(self, self.create_graph())
 
     @__server_connection
     def get_graph_ids(self):
@@ -797,7 +797,7 @@ def extract_subgraph(
         selection=None,
         edge_weight_property="",
         default_edge_weight=1.0,
-        allow_multi_edges=False,
+        check_multi_edges=True,
         renumber_graph=True,
         add_edge_data=True,
         graph_id=defaults.graph_id,
@@ -811,7 +811,7 @@ def extract_subgraph(
         create_using : string, default is None
             String describing the type of Graph object to create from the
             selected subgraph of vertices and edges. The default (None) results
-            in a cugraph.Graph object.
+            in a directed cugraph.MultiGraph object.
 
         selection : int, default is None
             A PropertySelection ID returned from one or more calls to
@@ -830,10 +830,10 @@ def extract_subgraph(
             The value to use when an edge property is specified but not present
             on an edge.
 
-        allow_multi_edges : bool
-            If True, multiple edges should be used to create the resulting
-            Graph, otherwise multiple edges will be detected and an exception
-            raised.
+        check_multi_edges : bool (default is True)
+            When True and create_using argument is given and not a MultiGraph,
+            this will perform an expensive check to verify that the edges in
+            the edge dataframe do not form a multigraph with duplicate edges.
 
         graph_id : int, default is defaults.graph_id
            The graph ID to extract the subgraph from. If the ID passed is not
@@ -861,7 +861,7 @@ def extract_subgraph(
             selection,
             edge_weight_property,
             default_edge_weight,
-            allow_multi_edges,
+            check_multi_edges,
             renumber_graph,
             add_edge_data,
             graph_id,
@@ -979,7 +979,7 @@ def get_graph_edge_data(
     def is_vertex_property(self, property_key, graph_id=defaults.graph_id):
         """
         Returns True if the given property key is for a valid vertex property
-        in the given graph, false otherwise.e
+        in the given graph, False otherwise.
 
         Parameters
         ----------