cugraph/cudf_interop/
egress.rs

1use cudf::{AggSpec, Exec, GroupByAgg, Runtime, TypeId, core::Column, frame::factorization::Codebook};
2
3use super::errors::map_cudf_error_with;
4use crate::{
5    centrality::{CentralityResult, EdgeBetweennessCentralityResult, HitsResult},
6    community::{
7        EcgResult, ExtractEgoResult, KTrussSubgraphResult, LeidenResult, LouvainResult, SpectralClusteringResult,
8        TriangleCountResult,
9    },
10    error::{Error, ErrorKind, Result},
11    labeling::LabelingResult,
12    layout::LayoutResult,
13    similarity::{
14        CosineSimilarityResult, JaccardSimilarityResult, OverlapSimilarityResult, SorensenSimilarityResult, VertexPairs,
15    },
16    structure::{
17        CoreNumberResult, KCoreResult,
18        degrees::{DegreeValues, DegreesResult, InDegreesResult, OutDegreesResult},
19        typed_results::{Distances, IntegerValues, ScoreValues, VertexIds},
20    },
21    traversal::{bfs::BfsResult, sssp::SsspResult},
22    tree::MinimumSpanningTreeResult,
23};
24
25#[cfg(test)]
26mod tests;
27
28fn series_from_column(column: Column, operation: &'static str) -> Result<cudf::Series> {
29    cudf::Series::from_column(column).map_err(|error| {
30        map_cudf_error_with(
31            error,
32            operation,
33            "failed to create cudf Series from cugraph result column",
34            std::iter::empty(),
35        )
36    })
37}
38
39fn series_from_column_result(column: Result<Column>, operation: &'static str) -> Result<cudf::Series> {
40    series_from_column(column?, operation)
41}
42
43pub(super) fn dataframe_from_named_series<I, S>(series: I, operation: &'static str) -> Result<cudf::DataFrame>
44where
45    I: IntoIterator<Item = (S, cudf::Series)>,
46    S: Into<String>,
47{
48    let series = series.into_iter();
49    let (series_count, _) = series.size_hint();
50    cudf::DataFrame::from_series(series).map_err(|error| {
51        map_cudf_error_with(
52            error,
53            operation,
54            "failed to build cudf DataFrame from cugraph result columns",
55            [("series_count", series_count.to_string())],
56        )
57    })
58}
59
60// ═══════════════════════════════════════════════════════════════════════════
61// to_series; result → cudf Series conversion
62// ═══════════════════════════════════════════════════════════════════════════
63
64impl VertexIds<'_> {
65    /// Copies this GPU array into an unnamed cudf [`Series`](cudf::Series)
66    /// bound to `runtime`.
67    ///
68    /// Device-to-device copy: unavoidable because the cugraph C API owns
69    /// result memory and frees it on drop. Matches Python cudf/cugraph behavior.
70    pub fn to_series(&self, runtime: &Runtime) -> Result<cudf::Series> {
71        series_from_column_result(
72            match self {
73                Self::I32(v) => v.raw_view().to_cudf_column(runtime),
74                Self::I64(v) => v.raw_view().to_cudf_column(runtime),
75            },
76            "VertexIds::to_series",
77        )
78    }
79}
80
81impl ScoreValues<'_> {
82    /// Copies this GPU array into an unnamed cudf [`Series`](cudf::Series)
83    /// bound to `runtime`.
84    ///
85    /// Device-to-device copy: unavoidable because the cugraph C API owns
86    /// result memory and frees it on drop. Matches Python cudf/cugraph behavior.
87    pub fn to_series(&self, runtime: &Runtime) -> Result<cudf::Series> {
88        series_from_column_result(
89            match self {
90                Self::F32(v) => v.raw_view().to_cudf_column(runtime),
91                Self::F64(v) => v.raw_view().to_cudf_column(runtime),
92            },
93            "ScoreValues::to_series",
94        )
95    }
96}
97
98impl Distances<'_> {
99    /// Copies this GPU array into an unnamed cudf [`Series`](cudf::Series)
100    /// bound to `runtime`.
101    ///
102    /// Device-to-device copy: unavoidable because the cugraph C API owns
103    /// result memory and frees it on drop. Matches Python cudf/cugraph behavior.
104    pub fn to_series(&self, runtime: &Runtime) -> Result<cudf::Series> {
105        series_from_column_result(
106            match self {
107                Self::I32(v) => v.raw_view().to_cudf_column(runtime),
108                Self::I64(v) => v.raw_view().to_cudf_column(runtime),
109                Self::F32(v) => v.raw_view().to_cudf_column(runtime),
110                Self::F64(v) => v.raw_view().to_cudf_column(runtime),
111            },
112            "Distances::to_series",
113        )
114    }
115}
116
117impl DegreeValues<'_> {
118    /// Copies this GPU array into an unnamed cudf [`Series`](cudf::Series)
119    /// bound to `runtime`.
120    ///
121    /// Device-to-device copy: unavoidable because the cugraph C API owns
122    /// result memory and frees it on drop. Matches Python cudf/cugraph behavior.
123    pub fn to_series(&self, runtime: &Runtime) -> Result<cudf::Series> {
124        series_from_column_result(
125            match self {
126                Self::I32(v) => v.raw_view().to_cudf_column(runtime),
127                Self::I64(v) => v.raw_view().to_cudf_column(runtime),
128            },
129            "DegreeValues::to_series",
130        )
131    }
132}
133
134impl IntegerValues<'_> {
135    /// Copies this GPU array into an unnamed cudf [`Series`](cudf::Series)
136    /// bound to `runtime`.
137    ///
138    /// Device-to-device copy: unavoidable because the cugraph C API owns
139    /// result memory and frees it on drop. Matches Python cudf/cugraph behavior.
140    pub fn to_series(&self, runtime: &Runtime) -> Result<cudf::Series> {
141        series_from_column_result(
142            match self {
143                Self::I32(v) => v.raw_view().to_cudf_column(runtime),
144                Self::I64(v) => v.raw_view().to_cudf_column(runtime),
145            },
146            "IntegerValues::to_series",
147        )
148    }
149}
150
151// ═══════════════════════════════════════════════════════════════════════════
152// TryFrom impls; result structs → cudf::DataFrame
153// ═══════════════════════════════════════════════════════════════════════════
154
155pub(crate) fn centrality_result_to_dataframe(result: &CentralityResult, runtime: &Runtime) -> Result<cudf::DataFrame> {
156    let operation = "centrality_result_to_dataframe";
157    dataframe_from_named_series(
158        [("vertex", result.vertex_ids()?.to_series(runtime)?), ("value", result.scores()?.to_series(runtime)?)],
159        operation,
160    )
161}
162
163pub(crate) fn hits_result_to_dataframe(result: &HitsResult, runtime: &Runtime) -> Result<cudf::DataFrame> {
164    let operation = "hits_result_to_dataframe";
165    dataframe_from_named_series(
166        [
167            ("vertex", result.vertex_ids()?.to_series(runtime)?),
168            ("hub", result.hub_scores()?.to_series(runtime)?),
169            ("authority", result.authority_scores()?.to_series(runtime)?),
170        ],
171        operation,
172    )
173}
174
175pub(crate) fn edge_betweenness_centrality_result_to_dataframe(
176    result: &EdgeBetweennessCentralityResult,
177    runtime: &Runtime,
178) -> Result<cudf::DataFrame> {
179    let operation = "edge_betweenness_centrality_result_to_dataframe";
180    let source_series = result.sources()?.to_series(runtime)?;
181    let destination_series = result.destinations()?.to_series(runtime)?;
182    let edge_id_series = result.edge_ids()?.map(|values| values.to_series(runtime)).transpose()?;
183    let score_series = result.scores()?.to_series(runtime)?;
184
185    let mut series = Vec::with_capacity(4);
186    series.push(("source", source_series));
187    series.push(("destination", destination_series));
188    if let Some(edge_id) = edge_id_series {
189        series.push(("edge_id", edge_id));
190    }
191    series.push(("score", score_series));
192
193    dataframe_from_named_series(series, operation)
194}
195
196pub(crate) fn louvain_result_to_dataframe(result: &LouvainResult, runtime: &Runtime) -> Result<cudf::DataFrame> {
197    let operation = "louvain_result_to_dataframe";
198    dataframe_from_named_series(
199        [("vertex", result.vertex_ids()?.to_series(runtime)?), ("partition", result.clusters()?.to_series(runtime)?)],
200        operation,
201    )
202}
203
204pub(crate) fn leiden_result_to_dataframe(result: &LeidenResult, runtime: &Runtime) -> Result<cudf::DataFrame> {
205    let operation = "leiden_result_to_dataframe";
206    dataframe_from_named_series(
207        [("vertex", result.vertex_ids()?.to_series(runtime)?), ("partition", result.clusters()?.to_series(runtime)?)],
208        operation,
209    )
210}
211
212pub(crate) fn ecg_result_to_dataframe(result: &EcgResult, runtime: &Runtime) -> Result<cudf::DataFrame> {
213    let operation = "ecg_result_to_dataframe";
214    dataframe_from_named_series(
215        [("vertex", result.vertex_ids()?.to_series(runtime)?), ("partition", result.clusters()?.to_series(runtime)?)],
216        operation,
217    )
218}
219
220pub(crate) fn spectral_clustering_result_to_dataframe(
221    result: &SpectralClusteringResult,
222    runtime: &Runtime,
223) -> Result<cudf::DataFrame> {
224    let operation = "spectral_clustering_result_to_dataframe";
225    dataframe_from_named_series(
226        [("vertex", result.vertex_ids()?.to_series(runtime)?), ("partition", result.clusters()?.to_series(runtime)?)],
227        operation,
228    )
229}
230
231pub(crate) fn bfs_result_to_dataframe(result: &BfsResult, runtime: &Runtime) -> Result<cudf::DataFrame> {
232    let operation = "bfs_result_to_dataframe";
233    let vertex_series = result.vertex_ids()?.to_series(runtime)?;
234    let distance_series = result.distances()?.to_series(runtime)?;
235
236    match result.predecessors()? {
237        Some(predecessors) => {
238            let predecessor_series = predecessors.to_series(runtime)?;
239            dataframe_from_named_series(
240                [("vertex", vertex_series), ("distance", distance_series), ("predecessor", predecessor_series)],
241                operation,
242            )
243        },
244        None => dataframe_from_named_series([("vertex", vertex_series), ("distance", distance_series)], operation),
245    }
246}
247
248pub(crate) fn sssp_result_to_dataframe(result: &SsspResult, runtime: &Runtime) -> Result<cudf::DataFrame> {
249    let operation = "sssp_result_to_dataframe";
250    let vertex_series = result.vertex_ids()?.to_series(runtime)?;
251    let distance_series = result.distances()?.to_series(runtime)?;
252
253    match result.predecessors()? {
254        Some(predecessors) => {
255            let predecessor_series = predecessors.to_series(runtime)?;
256            dataframe_from_named_series(
257                [("vertex", vertex_series), ("distance", distance_series), ("predecessor", predecessor_series)],
258                operation,
259            )
260        },
261        None => dataframe_from_named_series([("vertex", vertex_series), ("distance", distance_series)], operation),
262    }
263}
264
265pub(crate) fn layout_result_to_dataframe(result: &LayoutResult, runtime: &Runtime) -> Result<cudf::DataFrame> {
266    let operation = "layout_result_to_dataframe";
267    let vertex_series = result.vertex_ids()?.to_series(runtime)?;
268    let x_series = result.x()?.to_series(runtime)?;
269    let y_series = result.y()?.to_series(runtime)?;
270
271    dataframe_from_named_series([("vertex", vertex_series), ("x", x_series), ("y", y_series)], operation)
272}
273
274pub(crate) fn degrees_result_to_dataframe(result: &DegreesResult, runtime: &Runtime) -> Result<cudf::DataFrame> {
275    let operation = "degrees_result_to_dataframe";
276    dataframe_from_named_series(
277        [
278            ("vertex", result.vertex_ids()?.to_series(runtime)?),
279            ("in_degree", result.in_degrees()?.to_series(runtime)?),
280            ("out_degree", result.out_degrees()?.to_series(runtime)?),
281        ],
282        operation,
283    )
284}
285
286pub(crate) fn in_degrees_result_to_dataframe(result: &InDegreesResult, runtime: &Runtime) -> Result<cudf::DataFrame> {
287    let operation = "in_degrees_result_to_dataframe";
288    dataframe_from_named_series(
289        [("vertex", result.vertex_ids()?.to_series(runtime)?), ("in_degree", result.in_degrees()?.to_series(runtime)?)],
290        operation,
291    )
292}
293
294pub(crate) fn out_degrees_result_to_dataframe(result: &OutDegreesResult, runtime: &Runtime) -> Result<cudf::DataFrame> {
295    let operation = "out_degrees_result_to_dataframe";
296    dataframe_from_named_series(
297        [
298            ("vertex", result.vertex_ids()?.to_series(runtime)?),
299            ("out_degree", result.out_degrees()?.to_series(runtime)?),
300        ],
301        operation,
302    )
303}
304
305pub(crate) fn labeling_result_to_dataframe(result: &LabelingResult, runtime: &Runtime) -> Result<cudf::DataFrame> {
306    let operation = "labeling_result_to_dataframe";
307    let frame = dataframe_from_named_series(
308        [("vertex", result.vertex_ids()?.to_series(runtime)?), ("label", result.labels()?.to_series(runtime)?)],
309        operation,
310    )?;
311    canonicalize_component_labels(frame, operation)
312}
313
314/// Canonicalize a `("vertex", "label")` component frame by mapping each label
315/// to the minimum vertex id of its component.
316///
317/// Raw cuGraph component labels reflect non-deterministic kernel execution order,
318/// and weakly-connected components report labels in the internal renumbered space.
319/// Mapping each label to `MIN(vertex)` over its component produces deterministic,
320/// run-stable labels expressed in external vertex IDs while preserving row order.
321fn canonicalize_component_labels(frame: cudf::DataFrame, operation: &'static str) -> Result<cudf::DataFrame> {
322    if frame.num_rows() == 0 {
323        return Ok(frame);
324    }
325    for role in ["vertex", "label"] {
326        let dtype = frame.series(role).and_then(|series| series.dtype()).map_err(|error| {
327            map_cudf_error_with(error, operation, "failed to inspect component integer dtype", std::iter::empty())
328        })?;
329        if !matches!(dtype.id(), TypeId::Int32 | TypeId::Int64) {
330            return Err(Error::new(ErrorKind::UnsupportedType, operation)
331                .with_message(format!("unexpected component {role} dtype: {:?}", dtype.id())));
332        }
333    }
334    let mapping = frame
335        .groupby(["label"])
336        .on(Exec::RETAINED)
337        .agg(&[AggSpec::new("vertex", GroupByAgg::Min).with_alias("canonical")])
338        .map_err(|error| {
339            map_cudf_error_with(error, operation, "failed to compute component minimum vertex ids", std::iter::empty())
340        })?;
341    let mapping_keys = mapping.series("label").map_err(|error| {
342        map_cudf_error_with(error, operation, "failed to read component label mapping", std::iter::empty())
343    })?;
344    let codebook = Codebook::from_keys(mapping_keys).map_err(|error| {
345        map_cudf_error_with(error, operation, "failed to index component labels", std::iter::empty())
346    })?;
347    let label = frame.series("label").map_err(|error| {
348        map_cudf_error_with(error, operation, "failed to read component label column", std::iter::empty())
349    })?;
350    let codes = codebook.encode(&label, Exec::RETAINED).map_err(|error| {
351        map_cudf_error_with(error, operation, "failed to locate component minimum vertex ids", std::iter::empty())
352    })?;
353    // Gather requires non-null indices. i32::MAX exceeds the largest valid
354    // cuDF row index and preserves absent or null labels as null output rows.
355    let indices = codes.on(Exec::RETAINED).fillna(i32::MAX).map_err(|error| {
356        map_cudf_error_with(error, operation, "failed to preserve null component labels", std::iter::empty())
357    })?;
358    let canonical =
359        mapping.series("canonical").and_then(|values| values.on(Exec::RETAINED).gather(&indices)).map_err(|error| {
360            map_cudf_error_with(error, operation, "failed to gather canonical component labels", std::iter::empty())
361        })?;
362    let vertex = frame.series("vertex").map_err(|error| {
363        map_cudf_error_with(error, operation, "failed to read component vertex column", std::iter::empty())
364    })?;
365    dataframe_from_named_series([("vertex", vertex), ("label", canonical)], operation)
366}
367
368pub(crate) fn core_number_result_to_dataframe(result: &CoreNumberResult, runtime: &Runtime) -> Result<cudf::DataFrame> {
369    let operation = "core_number_result_to_dataframe";
370    dataframe_from_named_series(
371        [
372            ("vertex", result.vertex_ids()?.to_series(runtime)?),
373            ("core_number", result.core_numbers()?.to_series(runtime)?),
374        ],
375        operation,
376    )
377}
378
379pub(crate) fn triangle_count_result_to_dataframe(
380    result: &TriangleCountResult,
381    runtime: &Runtime,
382) -> Result<cudf::DataFrame> {
383    let operation = "triangle_count_result_to_dataframe";
384    dataframe_from_named_series(
385        [
386            ("vertex", result.vertex_ids()?.to_series(runtime)?),
387            ("triangle_count", result.counts()?.to_series(runtime)?),
388        ],
389        operation,
390    )
391}
392
393fn edge_result_to_dataframe(
394    sources: VertexIds<'_>,
395    destinations: VertexIds<'_>,
396    weights: Option<ScoreValues<'_>>,
397    edge_ids: Option<IntegerValues<'_>>,
398    edge_type_ids: Option<IntegerValues<'_>>,
399    runtime: &Runtime,
400    operation: &'static str,
401) -> Result<cudf::DataFrame> {
402    let source_series = sources.to_series(runtime)?;
403    let destination_series = destinations.to_series(runtime)?;
404    let weight_series = weights.map(|values| values.to_series(runtime)).transpose()?;
405    let edge_id_series = edge_ids.map(|values| values.to_series(runtime)).transpose()?;
406    let edge_type_series = edge_type_ids.map(|values| values.to_series(runtime)).transpose()?;
407
408    let mut series = Vec::with_capacity(5);
409    series.push(("source", source_series));
410    series.push(("destination", destination_series));
411    if let Some(weight) = weight_series {
412        series.push(("weight", weight));
413    }
414    if let Some(edge_id) = edge_id_series {
415        series.push(("edge_id", edge_id));
416    }
417    if let Some(edge_type) = edge_type_series {
418        series.push(("edge_type", edge_type));
419    }
420
421    dataframe_from_named_series(series, operation)
422}
423
424pub(crate) fn k_core_result_to_dataframe(result: &KCoreResult, runtime: &Runtime) -> Result<cudf::DataFrame> {
425    edge_result_to_dataframe(
426        result.sources()?,
427        result.destinations()?,
428        result.weights()?,
429        None,
430        None,
431        runtime,
432        "k_core_result_to_dataframe",
433    )
434}
435
436pub(crate) fn minimum_spanning_tree_result_to_dataframe(
437    result: &MinimumSpanningTreeResult,
438    runtime: &Runtime,
439) -> Result<cudf::DataFrame> {
440    edge_result_to_dataframe(
441        result.sources()?,
442        result.destinations()?,
443        result.weights()?,
444        result.edge_ids()?,
445        result.edge_type_ids()?,
446        runtime,
447        "minimum_spanning_tree_result_to_dataframe",
448    )
449}
450
451pub(crate) fn k_truss_subgraph_result_to_dataframe(
452    result: &KTrussSubgraphResult,
453    runtime: &Runtime,
454) -> Result<cudf::DataFrame> {
455    edge_result_to_dataframe(
456        result.sources()?,
457        result.destinations()?,
458        result.weights()?,
459        result.edge_ids()?,
460        result.edge_type_ids()?,
461        runtime,
462        "k_truss_subgraph_result_to_dataframe",
463    )
464}
465
466pub(crate) fn extract_ego_result_to_dataframe(result: &ExtractEgoResult, runtime: &Runtime) -> Result<cudf::DataFrame> {
467    edge_result_to_dataframe(
468        result.sources()?,
469        result.destinations()?,
470        result.weights()?,
471        result.edge_ids()?,
472        result.edge_type_ids()?,
473        runtime,
474        "extract_ego_result_to_dataframe",
475    )
476}
477
478pub(crate) fn vertex_pairs_to_dataframe(result: &VertexPairs, runtime: &Runtime) -> Result<cudf::DataFrame> {
479    let operation = "vertex_pairs_to_dataframe";
480    dataframe_from_named_series(
481        [("first", result.first()?.to_series(runtime)?), ("second", result.second()?.to_series(runtime)?)],
482        operation,
483    )
484}
485
486fn similarity_result_to_dataframe(
487    first: VertexIds<'_>,
488    second: VertexIds<'_>,
489    scores: ScoreValues<'_>,
490    runtime: &Runtime,
491    operation: &'static str,
492) -> Result<cudf::DataFrame> {
493    dataframe_from_named_series(
494        [
495            ("first", first.to_series(runtime)?),
496            ("second", second.to_series(runtime)?),
497            ("similarity", scores.to_series(runtime)?),
498        ],
499        operation,
500    )
501}
502
503pub(crate) fn jaccard_similarity_result_to_dataframe(
504    result: &JaccardSimilarityResult,
505    runtime: &Runtime,
506) -> Result<cudf::DataFrame> {
507    similarity_result_to_dataframe(
508        result.first()?,
509        result.second()?,
510        result.scores()?,
511        runtime,
512        "jaccard_similarity_result_to_dataframe",
513    )
514}
515
516pub(crate) fn sorensen_similarity_result_to_dataframe(
517    result: &SorensenSimilarityResult,
518    runtime: &Runtime,
519) -> Result<cudf::DataFrame> {
520    similarity_result_to_dataframe(
521        result.first()?,
522        result.second()?,
523        result.scores()?,
524        runtime,
525        "sorensen_similarity_result_to_dataframe",
526    )
527}
528
529pub(crate) fn overlap_similarity_result_to_dataframe(
530    result: &OverlapSimilarityResult,
531    runtime: &Runtime,
532) -> Result<cudf::DataFrame> {
533    similarity_result_to_dataframe(
534        result.first()?,
535        result.second()?,
536        result.scores()?,
537        runtime,
538        "overlap_similarity_result_to_dataframe",
539    )
540}
541
542pub(crate) fn cosine_similarity_result_to_dataframe(
543    result: &CosineSimilarityResult,
544    runtime: &Runtime,
545) -> Result<cudf::DataFrame> {
546    similarity_result_to_dataframe(
547        result.first()?,
548        result.second()?,
549        result.scores()?,
550        runtime,
551        "cosine_similarity_result_to_dataframe",
552    )
553}
cugraph/cudf_interop/egress.rs

cugraph/cudf_interop/
egress.rs
