Fixes for On-Going MG Test Failures

python/cugraph/cugraph/dask/community/induced_subgraph.py

@@ -1,4 +1,4 @@
-# Copyright (c) 2022-2023, NVIDIA CORPORATION.
+# Copyright (c) 2022-2024, NVIDIA CORPORATION.
 #
 # Licensed under the Apache License, Version 2.0 (the "License");
 # you may not use this file except in compliance with the License.
@@ -151,9 +151,9 @@ def induced_subgraph(
     # renumbered, the node ID must also be renumbered.
     if input_graph.renumbered:
         vertices = input_graph.lookup_internal_vertex_id(vertices)
-        vertices_type = input_graph.edgelist.edgelist_df.dtypes[0]
+        vertices_type = input_graph.edgelist.edgelist_df.dtypes.iloc[0]
     else:
-        vertices_type = input_graph.input_df.dtypes[0]
+        vertices_type = input_graph.input_df.dtypes.iloc[0]
 
     if isinstance(vertices, (cudf.Series, cudf.DataFrame)):
         vertices = dask_cudf.from_cudf(vertices, npartitions=input_graph._npartitions)

python/cugraph/cugraph/dask/cores/k_core.py

@@ -73,7 +73,7 @@ def k_core(input_graph, k=None, core_number=None, degree_type="bidirectional"):
         on input, output, or both directed edges, with valid values being
         "incoming", "outgoing", and "bidirectional" respectively.
 
-    core_number : cudf.DataFrame or das_cudf.DataFrame, optional (default=None)
+    core_number : cudf.DataFrame or dask_cudf.DataFrame, optional (default=None)
         Precomputed core number of the nodes of the graph G containing two
         cudf.Series of size V: the vertex identifiers and the corresponding
         core number values. If set to None, the core numbers of the nodes are
@@ -131,31 +131,25 @@ def k_core(input_graph, k=None, core_number=None, degree_type="bidirectional"):
         core_number = dcg.core_number(input_graph)
 
     if input_graph.renumbered is True:
-
         if len(input_graph.renumber_map.implementation.col_names) > 1:
             cols = core_number.columns[:-1].to_list()
         else:
             cols = "vertex"
-
             core_number = input_graph.add_internal_vertex_id(
                 core_number, "vertex", cols
             )
-
     if not isinstance(core_number, dask_cudf.DataFrame):
         if isinstance(core_number, cudf.DataFrame):
             # convert to dask_cudf in order to distribute the edges
             core_number = dask_cudf.from_cudf(core_number, input_graph._npartitions)
-
         else:
             raise TypeError(
                 f"'core_number' must be either None or of"
                 f"type cudf/dask_cudf, got: {type(core_number)}"
             )
-
     core_number = core_number.rename(columns={"core_number": "values"})
     if k is None:
         k = core_number["values"].max().compute()
-
     core_number = get_distributed_data(core_number)
     wait(core_number)
     core_number = core_number.worker_to_parts

python/cugraph/cugraph/dask/structure/mg_property_graph.py

@@ -1,4 +1,4 @@
-# Copyright (c) 2021-2023, NVIDIA CORPORATION.
+# Copyright (c) 2021-2024, NVIDIA CORPORATION.
 # 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
@@ -1123,8 +1123,9 @@ def fillna_edges(self, val=0):
             Series is passed, the index or keys are the columns to fill
             and the values are the fill value for the corresponding column.
         """
-
-        self.__edge_prop_dataframe = self.__edge_prop_dataframe.fillna(val).persist()
+        self.__edge_prop_dataframe["val"] = (
+            self.__edge_prop_dataframe["val"].fillna(val).persist()
+        )
 
     def select_vertices(self, expr, from_previous_selection=None):
         raise NotImplementedError

python/cugraph/cugraph/tests/core/test_core_number_mg.py

@@ -70,6 +70,7 @@ def test_sg_core_number(dask_client, dataset, degree_type, benchmark):
 @pytest.mark.parametrize("dataset", DATASETS)
 @pytest.mark.parametrize("degree_type", DEGREE_TYPE)
 def test_core_number(dask_client, dataset, degree_type, benchmark):
+    dataset.get_dask_edgelist(download=True)  # reload with MG edgelist
     dg = dataset.get_dask_graph(create_using=cugraph.Graph(directed=False))
 
     result_core_number = benchmark(dcg.core_number, dg, degree_type)

python/cugraph/cugraph/tests/core/test_k_core_mg.py

@@ -48,6 +48,7 @@ def setup_function():
 
 
 def get_sg_results(dataset, core_number, degree_type):
+    dataset.get_edgelist(download=True)
     G = dataset.get_graph(create_using=cugraph.Graph(directed=False))
 
     if core_number:
@@ -105,7 +106,7 @@ def test_dask_mg_k_core(dask_client, dataset, core_number, degree_type, benchmar
     expected_k_core_results, core_number = get_sg_results(
         dataset, core_number, degree_type
     )
-
+    dataset.get_dask_edgelist()  # reload with MG edgelist
     dg = dataset.get_dask_graph(create_using=cugraph.Graph(directed=False))
     k_core_results = benchmark(dcg.k_core, dg, core_number=core_number)
     k_core_results = (
@@ -123,6 +124,7 @@ def test_dask_mg_k_core(dask_client, dataset, core_number, degree_type, benchmar
 @pytest.mark.mg
 def test_dask_mg_k_core_invalid_input(dask_client):
     dataset = DATASETS[0]
+    dataset.get_dask_edgelist(download=True)  # reload with MG edgelist
     dg = dataset.get_dask_graph(create_using=cugraph.Graph(directed=True))
 
     with pytest.raises(ValueError):

python/cugraph/cugraph/tests/data_store/test_property_graph_mg.py

@@ -1042,7 +1042,8 @@ def test_add_data_noncontiguous(dask_client, set_index):
     for edge_type in ["cat", "dog", "pig"]:
         cur_df = df[df.edge_type == edge_type]
         if set_index:
-            cur_df = cur_df.set_index("vertex")
+            cur_df["ind_vertex"] = cur_df["vertex"]
+            cur_df = cur_df.set_index("ind_vertex")
         pG.add_vertex_data(cur_df, vertex_col_name="vertex", type_name=edge_type)
     for edge_type in ["cat", "dog", "pig"]:
         cur_df = pG.get_vertex_data(types=edge_type).compute()

python/cugraph/cugraph/tests/utils/test_replication_mg.py

@@ -1,4 +1,4 @@
-# Copyright (c) 2020-2023, NVIDIA CORPORATION.
+# Copyright (c) 2020-2024, NVIDIA CORPORATION.
 # 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
@@ -200,8 +200,8 @@ def test_enable_batch_edgelist_replication(graph_file, directed, dask_client):
     G = utils.generate_cugraph_graph_from_file(graph_file, directed)
     G.enable_batch()
     df = G.edgelist.edgelist_df
-    for worker in G.batch_edgelists:
-        replicated_df = G.batch_edgelists[worker].result()
+    for i in range(G.batch_edgelists.npartitions):
+        replicated_df = G.batch_edgelists.get_partition(i).compute()
         assert_frame_equal(df, replicated_df)