18443982156

Committed 12 Oct 2025 12:32PM UTC coverage: 48.052% (+0.2%) from 47.835%

Build # 18443982156

Build Type

Pull #152

github

Committed by

zommiommy

Commit Message

BatchCodec accepts Endianness

Pull Request Pull Request #152: Introduce BatchCodec to substitute BatchIterator

Run Details

112 of 233 new or added lines in 9 files covered. (48.07%)

7 existing lines in 3 files now uncovered.

4009 of 8343 relevant lines covered (48.05%)

21780055.91 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

23.53

/webgraph/src/transform/transpose.rs

/*
 * SPDX-FileCopyrightText: 2023 Inria
 * SPDX-FileCopyrightText: 2025 Sebastiano Vigna
 *
 * SPDX-License-Identifier: Apache-2.0 OR LGPL-2.1-or-later
 */

use crate::graphs::arc_list_graph;
use crate::labels::LeftIterator;
use crate::prelude::proj::Left;
use crate::prelude::sort_pairs::KMergeIters;
use crate::prelude::{LabeledSequentialGraph, SequentialGraph, SortPairs};
use crate::traits::graph::UnitLabelGraph;
use crate::traits::{NodeLabelsLender, SplitLabeling, UnitLender};
use crate::utils::{BatchCodec, CodecIter, DefaultBatchCodec, MemoryUsage, ParSortGraph};
use anyhow::Result;
use dsi_progress_logger::prelude::*;
use lender::prelude::*;
use tempfile::Builder;

/// Returns the transpose of the provided labeled graph as a [sequential
/// graph](crate::traits::SequentialGraph).
///
/// For the meaning of the additional parameters, see
/// [`SortPairs`](crate::prelude::sort_pairs::SortPairs).
#[allow(clippy::type_complexity)]
pub fn transpose_labeled<C: BatchCodec>(
    graph: &impl LabeledSequentialGraph<C::Label>,
    memory_usage: MemoryUsage,
    batch_codec: C,
) -> Result<arc_list_graph::ArcListGraph<KMergeIters<CodecIter<C>, C::Label>>>
where
    C::Label: Clone + 'static,
    CodecIter<C>: Clone + Send + Sync,
{
    let dir = Builder::new().prefix("transpose_").tempdir()?;
    let mut sorted = SortPairs::new_labeled(memory_usage, dir.path(), batch_codec)?;

    let mut pl = progress_logger![
        item_name = "node",
        expected_updates = Some(graph.num_nodes()),
        display_memory = true
    ];
    pl.start("Creating batches...");
    // create batches of sorted edges
    for_!( (src, succ) in graph.iter() {
        for (dst, l) in succ {
            sorted.push_labeled(dst, src, l)?;
        }
        pl.light_update();
    });
    // merge the batches
    let sorted = arc_list_graph::ArcListGraph::new_labeled(graph.num_nodes(), sorted.iter()?);
    pl.done();

    Ok(sorted)
}

/// Returns the transpose of the provided graph as a [sequential
/// graph](crate::traits::SequentialGraph).
///
/// For the meaning of the additional parameter, see
/// [`SortPairs`](crate::prelude::sort_pairs::SortPairs).
#[allow(clippy::type_complexity)]
pub fn transpose(
    graph: impl SequentialGraph,
    memory_usage: MemoryUsage,
) -> Result<Left<arc_list_graph::ArcListGraph<KMergeIters<CodecIter<DefaultBatchCodec>, ()>>>> {
    Ok(Left(transpose_labeled(
        &UnitLabelGraph(graph),
        memory_usage,
        DefaultBatchCodec::default(),
    )?))
}

/// Returns the transpose of the provided labeled graph as a [sequential
/// graph](crate::traits::SequentialGraph).
///
/// For the meaning of the additional parameters, see
/// [`SortPairs`](crate::prelude::sort_pairs::SortPairs).
#[allow(clippy::type_complexity)]
pub fn par_transpose_labeled<
    'graph,
    G: 'graph
        + LabeledSequentialGraph<C::Label>
        + for<'a> SplitLabeling<
            SplitLender<'a>: for<'b> NodeLabelsLender<
                'b,
                Label: crate::traits::Pair<Left = usize, Right = C::Label> + Copy,
                IntoIterator: IntoIterator<IntoIter: Send + Sync>,
            > + Send
                                 + Sync,
            IntoIterator<'a>: IntoIterator<IntoIter: Send + Sync>,
        >,
    C,
>(
    graph: &'graph G,
    memory_usage: MemoryUsage,
    batch_codec: C,
) -> Result<Vec<impl for<'a> NodeLabelsLender<'a, Label = (usize, C::Label)> + Send + Sync + 'graph>>
where
    C: BatchCodec + 'graph,
    CodecIter<C>: Clone + Send + Sync,
{
    let par_sort_graph = ParSortGraph::new(graph.num_nodes())?.memory_usage(memory_usage);
    let parts = num_cpus::get();

    let (start_nodes, pairs): (Vec<usize>, Vec<_>) = graph
        .split_iter(parts)
        .into_iter()
        .map(|(start_node, iter)| (start_node, iter.into_labeled_pairs()))
        .unzip();

    par_sort_graph
        .try_sort_labeled::<C, std::convert::Infallible>(batch_codec, pairs)?
        .into_iter()
        .enumerate()
        .map(|(i, res)| {
            arc_list_graph::Iter::try_new_from(graph.num_nodes(), res.into_iter(), start_nodes[i])
        })
        .collect()
}

pub fn par_transpose<'graph, G>(
    graph: &'graph G,
    memory_usage: MemoryUsage,
) -> Result<Vec<impl for<'a> NodeLabelsLender<'a, Label = usize> + Send + Sync>>
where
    // TODO check if 'graph is needed
    G: 'graph
        + SequentialGraph
        + for<'a> SplitLabeling<
            SplitLender<'a>: for<'b> NodeLabelsLender<
                'b,
                Label = usize,
                IntoIterator: IntoIterator<IntoIter: Send + Sync>,
            > + Send
                                 + Sync,
            IntoIterator<'a>: IntoIterator<IntoIter: Send + Sync>,
        >,
{
    let num_nodes = graph.num_nodes();
    let par_sort_graph = ParSortGraph::new(num_nodes)?.memory_usage(memory_usage);
    let parts = num_cpus::get();

    let (start_nodes, pairs): (Vec<usize>, Vec<_>) = graph
        .split_iter(parts)
        .into_iter()
        .map(|(start_node, iter)| (start_node, UnitLender(iter).into_labeled_pairs()))
        .unzip();

    let batch_codec = DefaultBatchCodec::default();
    Ok(par_sort_graph
        .try_sort_labeled::<_, std::convert::Infallible>(batch_codec, pairs)?
        .into_iter()
        .enumerate()
        .map(|(i, res)| {
            LeftIterator(
                arc_list_graph::Iter::try_new_from(
                    graph.num_nodes(),
                    res.into_iter(),
                    start_nodes[i],
                )
                .unwrap()
                .take(*start_nodes.get(i + 1).unwrap_or(&num_nodes) - start_nodes[i]),
            )
        })
        .collect::<Vec<_>>())
}

1	/*
2	* SPDX-FileCopyrightText: 2023 Inria
3	* SPDX-FileCopyrightText: 2025 Sebastiano Vigna
4	*
5	* SPDX-License-Identifier: Apache-2.0 OR LGPL-2.1-or-later
6	*/
7
8	use crate::graphs::arc_list_graph;
9	use crate::labels::LeftIterator;
10	use crate::prelude::proj::Left;
11	use crate::prelude::sort_pairs::KMergeIters;
12	use crate::prelude::{LabeledSequentialGraph, SequentialGraph, SortPairs};
13	use crate::traits::graph::UnitLabelGraph;
14	use crate::traits::{NodeLabelsLender, SplitLabeling, UnitLender};
15	use crate::utils::{BatchCodec, CodecIter, DefaultBatchCodec, MemoryUsage, ParSortGraph};
16	use anyhow::Result;
17	use dsi_progress_logger::prelude::*;
18	use lender::prelude::*;
19	use tempfile::Builder;
20
21	/// Returns the transpose of the provided labeled graph as a [sequential
22	/// graph](crate::traits::SequentialGraph).
23	///
24	/// For the meaning of the additional parameters, see
25	/// [`SortPairs`](crate::prelude::sort_pairs::SortPairs).
26	#[allow(clippy::type_complexity)]
27	pub fn transpose_labeled<C: BatchCodec>(	98✔
28	graph: &impl LabeledSequentialGraph<C::Label>,
29	memory_usage: MemoryUsage,
30	batch_codec: C,
31	) -> Result<arc_list_graph::ArcListGraph<KMergeIters<CodecIter<C>, C::Label>>>
32	where
33	C::Label: Clone + 'static,
34	CodecIter<C>: Clone + Send + Sync,
35	{
36	let dir = Builder::new().prefix("transpose_").tempdir()?;	392✔
37	let mut sorted = SortPairs::new_labeled(memory_usage, dir.path(), batch_codec)?;	98✔
38
39	let mut pl = progress_logger![	×
40	item_name = "node",	×
41	expected_updates = Some(graph.num_nodes()),	×
42	display_memory = true	×
43	];
44	pl.start("Creating batches...");	×
45	// create batches of sorted edges
46	for_!( (src, succ) in graph.iter() {	5,370✔
47	for (dst, l) in succ {	352,738✔
48	sorted.push_labeled(dst, src, l)?;	×
49	}
50	pl.light_update();	5,370✔
51	});
52	// merge the batches
53	let sorted = arc_list_graph::ArcListGraph::new_labeled(graph.num_nodes(), sorted.iter()?);	490✔
54	pl.done();	×
55
56	Ok(sorted)	×
57	}
58
59	/// Returns the transpose of the provided graph as a [sequential
60	/// graph](crate::traits::SequentialGraph).
61	///
62	/// For the meaning of the additional parameter, see
63	/// [`SortPairs`](crate::prelude::sort_pairs::SortPairs).
64	#[allow(clippy::type_complexity)]
65	pub fn transpose(	96✔
66	graph: impl SequentialGraph,
67	memory_usage: MemoryUsage,
68	) -> Result<Left<arc_list_graph::ArcListGraph<KMergeIters<CodecIter<DefaultBatchCodec>, ()>>>> {
69	Ok(Left(transpose_labeled(	96✔
70	&UnitLabelGraph(graph),	96✔
71	memory_usage,	96✔
72	DefaultBatchCodec::default(),	96✔
73	)?))
74	}
75
76	/// Returns the transpose of the provided labeled graph as a [sequential
77	/// graph](crate::traits::SequentialGraph).
78	///
79	/// For the meaning of the additional parameters, see
80	/// [`SortPairs`](crate::prelude::sort_pairs::SortPairs).
81	#[allow(clippy::type_complexity)]
82	pub fn par_transpose_labeled<	×
83	'graph,
84	G: 'graph
85	+ LabeledSequentialGraph<C::Label>
86	+ for<'a> SplitLabeling<
87	SplitLender<'a>: for<'b> NodeLabelsLender<
88	'b,
89	Label: crate::traits::Pair<Left = usize, Right = C::Label> + Copy,
90	IntoIterator: IntoIterator<IntoIter: Send + Sync>,
91	> + Send
92	+ Sync,
93	IntoIterator<'a>: IntoIterator<IntoIter: Send + Sync>,
94	>,
95	C,
96	>(
97	graph: &'graph G,
98	memory_usage: MemoryUsage,
99	batch_codec: C,
100	) -> Result<Vec<impl for<'a> NodeLabelsLender<'a, Label = (usize, C::Label)> + Send + Sync + 'graph>>
101	where
102	C: BatchCodec + 'graph,
103	CodecIter<C>: Clone + Send + Sync,
104	{
105	let par_sort_graph = ParSortGraph::new(graph.num_nodes())?.memory_usage(memory_usage);	×
106	let parts = num_cpus::get();	×
107
108	let (start_nodes, pairs): (Vec<usize>, Vec<_>) = graph	×
109	.split_iter(parts)	×
110	.into_iter()
111	.map(\|(start_node, iter)\| (start_node, iter.into_labeled_pairs()))	×
112	.unzip();
113
114	par_sort_graph	×
NEW 115	.try_sort_labeled::<C, std::convert::Infallible>(batch_codec, pairs)?	×
116	.into_iter()
117	.enumerate()
118	.map(\|(i, res)\| {	×
119	arc_list_graph::Iter::try_new_from(graph.num_nodes(), res.into_iter(), start_nodes[i])	×
120	})
121	.collect()
122	}
123
NEW 124	pub fn par_transpose<'graph, G>(	×
125	graph: &'graph G,
126	memory_usage: MemoryUsage,
127	) -> Result<Vec<impl for<'a> NodeLabelsLender<'a, Label = usize> + Send + Sync>>
128	where
129	// TODO check if 'graph is needed
130	G: 'graph
131	+ SequentialGraph
132	+ for<'a> SplitLabeling<
133	SplitLender<'a>: for<'b> NodeLabelsLender<
134	'b,
135	Label = usize,
136	IntoIterator: IntoIterator<IntoIter: Send + Sync>,
137	> + Send
138	+ Sync,
139	IntoIterator<'a>: IntoIterator<IntoIter: Send + Sync>,
140	>,
141	{
UNCOV 142	let num_nodes = graph.num_nodes();	×
143	let par_sort_graph = ParSortGraph::new(num_nodes)?.memory_usage(memory_usage);	×
144	let parts = num_cpus::get();	×
145
146	let (start_nodes, pairs): (Vec<usize>, Vec<_>) = graph	×
147	.split_iter(parts)	×
148	.into_iter()
149	.map(\|(start_node, iter)\| (start_node, UnitLender(iter).into_labeled_pairs()))	×
150	.unzip();
151
NEW 152	let batch_codec = DefaultBatchCodec::default();	×
153	Ok(par_sort_graph	×
NEW 154	.try_sort_labeled::<_, std::convert::Infallible>(batch_codec, pairs)?	×
155	.into_iter()	×
156	.enumerate()	×
157	.map(\|(i, res)\| {	×
158	LeftIterator(	×
159	arc_list_graph::Iter::try_new_from(	×
160	graph.num_nodes(),	×
161	res.into_iter(),	×
162	start_nodes[i],	×
163	)
164	.unwrap()	×
165	.take(*start_nodes.get(i + 1).unwrap_or(&num_nodes) - start_nodes[i]),	×
166	)
167	})
168	.collect::<Vec<_>>())	×
169	}

vigna / webgraph-rs / 18443982156

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous