19278973738

Committed 11 Nov 2025 09:25PM UTC coverage: 62.257% (+0.6%) from 61.674%

Build # 19278973738

Build Type

Pull #152

github

Committed by

web-flow

Commit Message

Merge d8e872162 into fedf79d67

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

Run Details

206 of 257 new or added lines in 10 files covered. (80.16%)

48 existing lines in 3 files now uncovered.

5247 of 8428 relevant lines covered (62.26%)

29389328.1 hits per line

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

91.3

/webgraph/src/utils/batch_codec/grouped_gaps.rs

/*
 * SPDX-FileCopyrightText: 2025 Tommaso Fontana
 *
 * SPDX-License-Identifier: Apache-2.0 OR LGPL-2.1-or-later
 */

use super::{BitReader, BitWriter};
use crate::traits::SortedIterator;
use crate::utils::{ArcMmapHelper, MmapHelper, Triple};
use crate::{
    traits::{BitDeserializer, BitSerializer},
    utils::{humanize, BatchCodec},
};

use std::sync::Arc;

use anyhow::{Context, Result};
use dsi_bitstream::prelude::*;
use mmap_rs::MmapFlags;
use rdst::*;

#[derive(Clone, Debug)]
/// A codec for encoding and decoding batches of triples using grouped gap compression.
///
/// This codec encodes triples of the form `(src, dst, label)` by grouping edges
/// with the same source node, and encoding the gaps between consecutive sources
/// and destinations using a specified code (default: gamma). The outdegree
/// (number of edges for each source) is also encoded using the specified code.
///
/// # Type Parameters
///
/// - `S`: Serializer for the labels, implementing [`BitSerializer`] for the label type.
/// - `D`: Deserializer for the labels, implementing [`BitDeserializer`] for the label type.
/// - `OUTDEGREE_CODE`: Code used for encoding outdegrees (default: [ɣ](dsi_bitstream::codes::gamma)).
/// - `SRC_CODE`: Code used for encoding source gaps (default: [ɣ](dsi_bitstream::codes::gamma)).
/// - `DST_CODE`: Code used for encoding destination gaps (default: [ɣ](dsi_bitstream::codes::gamma)).
///
/// # Encoding Format
///
/// 1. The batch length is written using delta coding.
/// 2. For each group of triples with the same source:
///     - The gap from the previous source is encoded.
///     - The outdegree (number of edges for this source) is encoded.
///     - For each destination:
///         - The gap from the previous destination is encoded.
///         - The label is serialized.
///
/// The bit deserializer must be [`Clone`] because we need one for each
/// [`GroupedGapsIterator`], and there are possible scenarios in which the
/// deserializer might be stateful.
pub struct GroupedGapsCodec<
    E: Endianness = NE,
    S: BitSerializer<E, BitWriter<E>> = (),
    D: BitDeserializer<E, BitReader<E>, DeserType = S::SerType> + Clone = (),
    const OUTDEGREE_CODE: usize = { dsi_bitstream::dispatch::code_consts::GAMMA },
    const SRC_CODE: usize = { dsi_bitstream::dispatch::code_consts::GAMMA },
    const DST_CODE: usize = { dsi_bitstream::dispatch::code_consts::DELTA },
> where
    BitReader<E>: BitRead<E>,
    BitWriter<E>: BitWrite<E>,
{
    /// Serializer for the labels.
    pub serializer: S,
    /// Deserializer for the labels.
    pub deserializer: D,

    pub _marker: core::marker::PhantomData<E>,
}

impl<E, S, D, const OUTDEGREE_CODE: usize, const SRC_CODE: usize, const DST_CODE: usize>
    GroupedGapsCodec<E, S, D, OUTDEGREE_CODE, SRC_CODE, DST_CODE>
where
    E: Endianness,
    S: BitSerializer<E, BitWriter<E>> + Send + Sync,
    D: BitDeserializer<E, BitReader<E>, DeserType = S::SerType> + Send + Sync + Clone,
    BitReader<E>: BitRead<E>,
    BitWriter<E>: BitWrite<E>,
{
    /// Creates a new `GroupedGapsCodec` with the given serializer and deserializer.
    pub fn new(serializer: S, deserializer: D) -> Self {
        Self {
            serializer,
            deserializer,
            _marker: core::marker::PhantomData,
        }
    }
}

impl<
        E: Endianness,
        S: BitSerializer<E, BitWriter<E>> + Default,
        D: BitDeserializer<E, BitReader<E>, DeserType = S::SerType> + Clone + Default,
        const OUTDEGREE_CODE: usize,
        const SRC_CODE: usize,
        const DST_CODE: usize,
    > Default for GroupedGapsCodec<E, S, D, OUTDEGREE_CODE, SRC_CODE, DST_CODE>
where
    BitReader<E>: BitRead<E>,
    BitWriter<E>: BitWrite<E>,
{
    fn default() -> Self {
        Self {
            serializer: S::default(),
            deserializer: D::default(),
            _marker: core::marker::PhantomData,
        }
    }
}

#[derive(Debug, Clone, Copy)]
/// Statistics about the encoding performed by [`GapsCodec`].
pub struct GroupedGapsStats {
    /// Total number of triples encoded
    pub total_triples: usize,
    /// Number of bits used for outdegrees
    pub outdegree_bits: usize,
    /// Number of bits used for source gaps
    pub src_bits: usize,
    //// Number of bits used for destination gaps
    pub dst_bits: usize,
    /// Number of bits used for labels
    pub labels_bits: usize,
}

impl core::fmt::Display for GroupedGapsStats {
    fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
        write!(
            f,
            "outdegree: {}B ({:.3} bits / arc), src: {}B ({:.3} bits / arc), dst: {}B ({:.3} bits / arc), labels: {}B ({:.3} bits / arc)",
            humanize(self.outdegree_bits as f64 / 8.0),
            self.outdegree_bits as f64 / self.total_triples as f64,
            humanize(self.src_bits as f64 / 8.0),
            self.src_bits as f64 / self.total_triples as f64,
            humanize(self.dst_bits as f64 / 8.0),
            self.dst_bits as f64 / self.total_triples as f64,
            humanize(self.labels_bits as f64 / 8.0),
            self.labels_bits as f64 / self.total_triples as f64,
        )
    }
}

impl<E, S, D, const OUTDEGREE_CODE: usize, const SRC_CODE: usize, const DST_CODE: usize> BatchCodec
    for GroupedGapsCodec<E, S, D, OUTDEGREE_CODE, SRC_CODE, DST_CODE>
where
    E: Endianness,
    S: BitSerializer<E, BitWriter<E>> + Send + Sync,
    D: BitDeserializer<E, BitReader<E>, DeserType = S::SerType> + Send + Sync + Clone,
    S::SerType: Send + Sync + Copy + 'static, // needed by radix sort
    BitReader<E>: BitRead<E> + CodesRead<E>,
    BitWriter<E>: BitWrite<E> + CodesWrite<E>,
{
    type Label = S::SerType;
    type DecodedBatch = GroupedGapsIterator<E, D, OUTDEGREE_CODE, SRC_CODE, DST_CODE>;
    type EncodedBatchStats = GroupedGapsStats;

    fn encode_batch(
        &self,
        path: impl AsRef<std::path::Path>,
        batch: &mut [((usize, usize), Self::Label)],
    ) -> Result<(usize, Self::EncodedBatchStats)> {
        let start = std::time::Instant::now();
        Triple::cast_batch_mut(batch).radix_sort_unstable();
        log::debug!("Sorted {} arcs in {:?}", batch.len(), start.elapsed());
        self.encode_sorted_batch(path, batch)
    }

    fn encode_sorted_batch(
        &self,
        path: impl AsRef<std::path::Path>,
        batch: &[((usize, usize), Self::Label)],
    ) -> Result<(usize, Self::EncodedBatchStats)> {
        debug_assert!(Triple::cast_batch(batch).is_sorted(), "Batch is not sorted");
        // create a batch file where to dump
        let file_path = path.as_ref();
        let file = std::io::BufWriter::with_capacity(
            1 << 16,
            std::fs::File::create(file_path).with_context(|| {
                format!(
                    "Could not create BatchIterator temporary file {}",
                    file_path.display()
                )
            })?,
        );
        // create a bitstream to write to the file
        let mut stream = <BufBitWriter<E, _>>::new(<WordAdapter<usize, _>>::new(file));

        // prefix the stream with the length of the batch
        // we use a delta code since it'll be a big number most of the time
        stream
            .write_delta(batch.len() as u64)
            .context("Could not write length")?;

        let mut stats = GroupedGapsStats {
            total_triples: batch.len(),
            outdegree_bits: 0,
            src_bits: 0,
            dst_bits: 0,
            labels_bits: 0,
        };
        // dump the triples to the bitstream
        let mut prev_src = 0;
        let mut i = 0;
        while i < batch.len() {
            let ((src, _), _) = batch[i];
            // write the source gap as gamma
            stats.src_bits += ConstCode::<SRC_CODE>
                .write(&mut stream, (src - prev_src) as _)
                .with_context(|| format!("Could not write {src} after {prev_src}"))?;
            // figure out how many edges have this source
            let outdegree = batch[i..].iter().take_while(|t| t.0 .0 == src).count();
            // write the outdegree
            stats.outdegree_bits += ConstCode::<OUTDEGREE_CODE>
                .write(&mut stream, outdegree as _)
                .with_context(|| format!("Could not write outdegree {outdegree} for {src}"))?;

            // encode the destinations
            let mut prev_dst = 0;
            for _ in 0..outdegree {
                let ((_, dst), label) = &batch[i];
                // write the destination gap as gamma
                stats.dst_bits += ConstCode::<DST_CODE>
                    .write(&mut stream, (dst - prev_dst) as _)
                    .with_context(|| format!("Could not write {dst} after {prev_dst}"))?;
                // write the label
                stats.labels_bits += self
                    .serializer
                    .serialize(label, &mut stream)
                    .context("Could not serialize label")?;
                prev_dst = *dst;
                i += 1;
            }
            prev_src = src;
        }
        // flush the stream and reset the buffer
        stream.flush().context("Could not flush stream")?;

        let total_bits = stats.outdegree_bits + stats.src_bits + stats.dst_bits + stats.labels_bits;
        Ok((total_bits, stats))
    }

    fn decode_batch(&self, path: impl AsRef<std::path::Path>) -> Result<Self::DecodedBatch> {
        // open the file
        let mut stream = <BufBitReader<E, _>>::new(MemWordReader::new(ArcMmapHelper(Arc::new(
            MmapHelper::mmap(
                path.as_ref(),
                MmapFlags::TRANSPARENT_HUGE_PAGES | MmapFlags::SEQUENTIAL,
            )
            .with_context(|| format!("Could not mmap {}", path.as_ref().display()))?,
        ))));

        // read the length of the batch (first value in the stream)
        let len = stream.read_delta().context("Could not read length")? as usize;

        // create the iterator
        Ok(GroupedGapsIterator {
            deserializer: self.deserializer.clone(),
            stream,
            len,
            current: 0,
            src: 0,
            dst_left: 0,
            prev_dst: 0,
        })
    }
}

#[derive(Clone, Debug)]
/// An iterator over triples encoded with gaps, this is returned by [`GroupedGapsCodec`].
pub struct GroupedGapsIterator<
    E: Endianness = NE,
    D: BitDeserializer<E, BitReader<E>> = (),
    const OUTDEGREE_CODE: usize = { dsi_bitstream::dispatch::code_consts::GAMMA },
    const SRC_CODE: usize = { dsi_bitstream::dispatch::code_consts::GAMMA },
    const DST_CODE: usize = { dsi_bitstream::dispatch::code_consts::GAMMA },
> where
    BitReader<E>: BitRead<E>,
    BitWriter<E>: BitWrite<E>,
{
    /// Deserializer for the labels
    deserializer: D,
    /// Bitstream to read from
    stream: BitReader<E>,
    /// Length of the iterator (number of triples)
    len: usize,
    /// Current position in the iterator
    current: usize,
    /// Current source node
    src: usize,
    /// Number of destinations left for the current source
    dst_left: usize,
    /// Previous destination node
    prev_dst: usize,
}

unsafe impl<
        E: Endianness,
        D: BitDeserializer<E, BitReader<E>>,
        const OUTDEGREE_CODE: usize,
        const SRC_CODE: usize,
        const DST_CODE: usize,
    > SortedIterator for GroupedGapsIterator<E, D, OUTDEGREE_CODE, SRC_CODE, DST_CODE>
where
    BitReader<E>: BitRead<E> + CodesRead<E>,
    BitWriter<E>: BitWrite<E> + CodesWrite<E>,
{
}

impl<
        E: Endianness,
        D: BitDeserializer<E, BitReader<E>>,
        const OUTDEGREE_CODE: usize,
        const SRC_CODE: usize,
        const DST_CODE: usize,
    > Iterator for GroupedGapsIterator<E, D, OUTDEGREE_CODE, SRC_CODE, DST_CODE>
where
    BitReader<E>: BitRead<E> + CodesRead<E>,
    BitWriter<E>: BitWrite<E> + CodesWrite<E>,
{
    type Item = ((usize, usize), D::DeserType);
    fn next(&mut self) -> Option<Self::Item> {
        if self.current >= self.len {
            return None;
        }
        if self.dst_left == 0 {
            // read a new source
            let src_gap = ConstCode::<SRC_CODE>.read(&mut self.stream).ok()?;
            self.src += src_gap as usize;
            // read the outdegree
            self.dst_left = ConstCode::<OUTDEGREE_CODE>.read(&mut self.stream).ok()? as usize;
            self.prev_dst = 0;
        }

        let dst_gap = ConstCode::<DST_CODE>.read(&mut self.stream).ok()?;
        let label = self.deserializer.deserialize(&mut self.stream).ok()?;
        self.prev_dst += dst_gap as usize;
        self.current += 1;
        self.dst_left -= 1;
        Some(((self.src, self.prev_dst), label))
    }

    fn size_hint(&self) -> (usize, Option<usize>) {
        (self.len(), Some(self.len()))
    }
}

impl<
        E: Endianness,
        D: BitDeserializer<E, BitReader<E>>,
        const OUTDEGREE_CODE: usize,
        const SRC_CODE: usize,
        const DST_CODE: usize,
    > ExactSizeIterator for GroupedGapsIterator<E, D, OUTDEGREE_CODE, SRC_CODE, DST_CODE>
where
    BitReader<E>: BitRead<E> + CodesRead<E>,
    BitWriter<E>: BitWrite<E> + CodesWrite<E>,
{
    fn len(&self) -> usize {
        self.len - self.current
    }
}

1	/*
2	* SPDX-FileCopyrightText: 2025 Tommaso Fontana
3	*
4	* SPDX-License-Identifier: Apache-2.0 OR LGPL-2.1-or-later
5	*/
6
7	use super::{BitReader, BitWriter};
8	use crate::traits::SortedIterator;
9	use crate::utils::{ArcMmapHelper, MmapHelper, Triple};
10	use crate::{
11	traits::{BitDeserializer, BitSerializer},
12	utils::{humanize, BatchCodec},
13	};
14
15	use std::sync::Arc;
16
17	use anyhow::{Context, Result};
18	use dsi_bitstream::prelude::*;
19	use mmap_rs::MmapFlags;
20	use rdst::*;
21
22	#[derive(Clone, Debug)]
23	/// A codec for encoding and decoding batches of triples using grouped gap compression.
24	///
25	/// This codec encodes triples of the form `(src, dst, label)` by grouping edges
26	/// with the same source node, and encoding the gaps between consecutive sources
27	/// and destinations using a specified code (default: gamma). The outdegree
28	/// (number of edges for each source) is also encoded using the specified code.
29	///
30	/// # Type Parameters
31	///
32	/// - `S`: Serializer for the labels, implementing [`BitSerializer`] for the label type.
33	/// - `D`: Deserializer for the labels, implementing [`BitDeserializer`] for the label type.
34	/// - `OUTDEGREE_CODE`: Code used for encoding outdegrees (default: [ɣ](dsi_bitstream::codes::gamma)).
35	/// - `SRC_CODE`: Code used for encoding source gaps (default: [ɣ](dsi_bitstream::codes::gamma)).
36	/// - `DST_CODE`: Code used for encoding destination gaps (default: [ɣ](dsi_bitstream::codes::gamma)).
37	///
38	/// # Encoding Format
39	///
40	/// 1. The batch length is written using delta coding.
41	/// 2. For each group of triples with the same source:
42	/// - The gap from the previous source is encoded.
43	/// - The outdegree (number of edges for this source) is encoded.
44	/// - For each destination:
45	/// - The gap from the previous destination is encoded.
46	/// - The label is serialized.
47	///
48	/// The bit deserializer must be [`Clone`] because we need one for each
49	/// [`GroupedGapsIterator`], and there are possible scenarios in which the
50	/// deserializer might be stateful.
51	pub struct GroupedGapsCodec<
52	E: Endianness = NE,
53	S: BitSerializer<E, BitWriter<E>> = (),
54	D: BitDeserializer<E, BitReader<E>, DeserType = S::SerType> + Clone = (),
55	const OUTDEGREE_CODE: usize = { dsi_bitstream::dispatch::code_consts::GAMMA },
56	const SRC_CODE: usize = { dsi_bitstream::dispatch::code_consts::GAMMA },
57	const DST_CODE: usize = { dsi_bitstream::dispatch::code_consts::DELTA },
58	> where
59	BitReader<E>: BitRead<E>,
60	BitWriter<E>: BitWrite<E>,
61	{
62	/// Serializer for the labels.
63	pub serializer: S,
64	/// Deserializer for the labels.
65	pub deserializer: D,
66
67	pub _marker: core::marker::PhantomData<E>,
68	}
69
70	impl<E, S, D, const OUTDEGREE_CODE: usize, const SRC_CODE: usize, const DST_CODE: usize>
71	GroupedGapsCodec<E, S, D, OUTDEGREE_CODE, SRC_CODE, DST_CODE>
72	where
73	E: Endianness,
74	S: BitSerializer<E, BitWriter<E>> + Send + Sync,
75	D: BitDeserializer<E, BitReader<E>, DeserType = S::SerType> + Send + Sync + Clone,
76	BitReader<E>: BitRead<E>,
77	BitWriter<E>: BitWrite<E>,
78	{
79	/// Creates a new `GroupedGapsCodec` with the given serializer and deserializer.
NEW 80	pub fn new(serializer: S, deserializer: D) -> Self {	×
81	Self {
82	serializer,
83	deserializer,
84	_marker: core::marker::PhantomData,
85	}
86	}
87	}
88
89	impl<
90	E: Endianness,
91	S: BitSerializer<E, BitWriter<E>> + Default,
92	D: BitDeserializer<E, BitReader<E>, DeserType = S::SerType> + Clone + Default,
93	const OUTDEGREE_CODE: usize,
94	const SRC_CODE: usize,
95	const DST_CODE: usize,
96	> Default for GroupedGapsCodec<E, S, D, OUTDEGREE_CODE, SRC_CODE, DST_CODE>
97	where
98	BitReader<E>: BitRead<E>,
99	BitWriter<E>: BitWrite<E>,
100	{
101	fn default() -> Self {	131✔
102	Self {
103	serializer: S::default(),	262✔
104	deserializer: D::default(),	131✔
105	_marker: core::marker::PhantomData,
106	}
107	}
108	}
109
110	#[derive(Debug, Clone, Copy)]
111	/// Statistics about the encoding performed by [`GapsCodec`].
112	pub struct GroupedGapsStats {
113	/// Total number of triples encoded
114	pub total_triples: usize,
115	/// Number of bits used for outdegrees
116	pub outdegree_bits: usize,
117	/// Number of bits used for source gaps
118	pub src_bits: usize,
119	//// Number of bits used for destination gaps
120	pub dst_bits: usize,
121	/// Number of bits used for labels
122	pub labels_bits: usize,
123	}
124
125	impl core::fmt::Display for GroupedGapsStats {
126	fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {	32✔
127	write!(	32✔
128	f,	32✔
129	"outdegree: {}B ({:.3} bits / arc), src: {}B ({:.3} bits / arc), dst: {}B ({:.3} bits / arc), labels: {}B ({:.3} bits / arc)",
130	humanize(self.outdegree_bits as f64 / 8.0),	64✔
131	self.outdegree_bits as f64 / self.total_triples as f64,	32✔
132	humanize(self.src_bits as f64 / 8.0),	64✔
133	self.src_bits as f64 / self.total_triples as f64,	32✔
134	humanize(self.dst_bits as f64 / 8.0),	64✔
135	self.dst_bits as f64 / self.total_triples as f64,	32✔
136	humanize(self.labels_bits as f64 / 8.0),	64✔
137	self.labels_bits as f64 / self.total_triples as f64,	32✔
138	)
139	}
140	}
141
142	impl<E, S, D, const OUTDEGREE_CODE: usize, const SRC_CODE: usize, const DST_CODE: usize> BatchCodec
143	for GroupedGapsCodec<E, S, D, OUTDEGREE_CODE, SRC_CODE, DST_CODE>
144	where
145	E: Endianness,
146	S: BitSerializer<E, BitWriter<E>> + Send + Sync,
147	D: BitDeserializer<E, BitReader<E>, DeserType = S::SerType> + Send + Sync + Clone,
148	S::SerType: Send + Sync + Copy + 'static, // needed by radix sort
149	BitReader<E>: BitRead<E> + CodesRead<E>,
150	BitWriter<E>: BitWrite<E> + CodesWrite<E>,
151	{
152	type Label = S::SerType;
153	type DecodedBatch = GroupedGapsIterator<E, D, OUTDEGREE_CODE, SRC_CODE, DST_CODE>;
154	type EncodedBatchStats = GroupedGapsStats;
155
156	fn encode_batch(	459✔
157	&self,
158	path: impl AsRef<std::path::Path>,
159	batch: &mut [((usize, usize), Self::Label)],
160	) -> Result<(usize, Self::EncodedBatchStats)> {
161	let start = std::time::Instant::now();	918✔
162	Triple::cast_batch_mut(batch).radix_sort_unstable();	918✔
163	log::debug!("Sorted {} arcs in {:?}", batch.len(), start.elapsed());	459✔
164	self.encode_sorted_batch(path, batch)	1,836✔
165	}
166
167	fn encode_sorted_batch(	459✔
168	&self,
169	path: impl AsRef<std::path::Path>,
170	batch: &[((usize, usize), Self::Label)],
171	) -> Result<(usize, Self::EncodedBatchStats)> {
172	debug_assert!(Triple::cast_batch(batch).is_sorted(), "Batch is not sorted");	1,377✔
173	// create a batch file where to dump
174	let file_path = path.as_ref();	1,377✔
175	let file = std::io::BufWriter::with_capacity(
176	1 << 16,	459✔
177	std::fs::File::create(file_path).with_context(\|\| {	1,377✔
NEW 178	format!(	×
NEW 179	"Could not create BatchIterator temporary file {}",	×
NEW 180	file_path.display()	×
181	)
182	})?,
183	);
184	// create a bitstream to write to the file
185	let mut stream = <BufBitWriter<E, _>>::new(<WordAdapter<usize, _>>::new(file));	1,836✔
186
187	// prefix the stream with the length of the batch
188	// we use a delta code since it'll be a big number most of the time
189	stream	459✔
190	.write_delta(batch.len() as u64)	918✔
191	.context("Could not write length")?;
192
193	let mut stats = GroupedGapsStats {
194	total_triples: batch.len(),	459✔
195	outdegree_bits: 0,
196	src_bits: 0,
197	dst_bits: 0,
198	labels_bits: 0,
199	};
200	// dump the triples to the bitstream
201	let mut prev_src = 0;	918✔
202	let mut i = 0;	918✔
203	while i < batch.len() {	4,726,460✔
204	let ((src, _), _) = batch[i];	4,725,542✔
205	// write the source gap as gamma
206	stats.src_bits += ConstCode::<SRC_CODE>	2,362,771✔
207	.write(&mut stream, (src - prev_src) as _)	7,088,313✔
208	.with_context(\|\| format!("Could not write {src} after {prev_src}"))?;	2,362,771✔
209	// figure out how many edges have this source
210	let outdegree = batch[i..].iter().take_while(\|t\| t.0 .0 == src).count();	93,710,435✔
211	// write the outdegree
212	stats.outdegree_bits += ConstCode::<OUTDEGREE_CODE>	2,362,771✔
213	.write(&mut stream, outdegree as _)	7,088,313✔
214	.with_context(\|\| format!("Could not write outdegree {outdegree} for {src}"))?;	2,362,771✔
215
216	// encode the destinations
217	let mut prev_dst = 0;	4,725,542✔
218	for _ in 0..outdegree {	2,362,771✔
219	let ((_, dst), label) = &batch[i];	115,757,895✔
220	// write the destination gap as gamma
221	stats.dst_bits += ConstCode::<DST_CODE>	38,585,965✔
222	.write(&mut stream, (dst - prev_dst) as _)	115,757,895✔
223	.with_context(\|\| format!("Could not write {dst} after {prev_dst}"))?;	38,585,965✔
224	// write the label
225	stats.labels_bits += self	38,585,965✔
226	.serializer	38,585,965✔
227	.serialize(label, &mut stream)	115,757,895✔
228	.context("Could not serialize label")?;	38,585,965✔
229	prev_dst = *dst;	38,585,965✔
230	i += 1;	38,585,965✔
231	}
232	prev_src = src;	2,362,771✔
233	}
234	// flush the stream and reset the buffer
235	stream.flush().context("Could not flush stream")?;	1,377✔
236
237	let total_bits = stats.outdegree_bits + stats.src_bits + stats.dst_bits + stats.labels_bits;	918✔
238	Ok((total_bits, stats))	459✔
239	}
240
241	fn decode_batch(&self, path: impl AsRef<std::path::Path>) -> Result<Self::DecodedBatch> {	162✔
242	// open the file
243	let mut stream = <BufBitReader<E, _>>::new(MemWordReader::new(ArcMmapHelper(Arc::new(	648✔
244	MmapHelper::mmap(	162✔
245	path.as_ref(),	324✔
246	MmapFlags::TRANSPARENT_HUGE_PAGES \| MmapFlags::SEQUENTIAL,	162✔
247	)
248	.with_context(\|\| format!("Could not mmap {}", path.as_ref().display()))?,	162✔
249	))));
250
251	// read the length of the batch (first value in the stream)
252	let len = stream.read_delta().context("Could not read length")? as usize;	648✔
253
254	// create the iterator
255	Ok(GroupedGapsIterator {	162✔
256	deserializer: self.deserializer.clone(),	486✔
257	stream,	162✔
258	len,	162✔
259	current: 0,	162✔
260	src: 0,	162✔
261	dst_left: 0,	162✔
262	prev_dst: 0,	162✔
263	})
264	}
265	}
266
267	#[derive(Clone, Debug)]
268	/// An iterator over triples encoded with gaps, this is returned by [`GroupedGapsCodec`].
269	pub struct GroupedGapsIterator<
270	E: Endianness = NE,
271	D: BitDeserializer<E, BitReader<E>> = (),
272	const OUTDEGREE_CODE: usize = { dsi_bitstream::dispatch::code_consts::GAMMA },
273	const SRC_CODE: usize = { dsi_bitstream::dispatch::code_consts::GAMMA },
274	const DST_CODE: usize = { dsi_bitstream::dispatch::code_consts::GAMMA },
275	> where
276	BitReader<E>: BitRead<E>,
277	BitWriter<E>: BitWrite<E>,
278	{
279	/// Deserializer for the labels
280	deserializer: D,
281	/// Bitstream to read from
282	stream: BitReader<E>,
283	/// Length of the iterator (number of triples)
284	len: usize,
285	/// Current position in the iterator
286	current: usize,
287	/// Current source node
288	src: usize,
289	/// Number of destinations left for the current source
290	dst_left: usize,
291	/// Previous destination node
292	prev_dst: usize,
293	}
294
295	unsafe impl<
296	E: Endianness,
297	D: BitDeserializer<E, BitReader<E>>,
298	const OUTDEGREE_CODE: usize,
299	const SRC_CODE: usize,
300	const DST_CODE: usize,
301	> SortedIterator for GroupedGapsIterator<E, D, OUTDEGREE_CODE, SRC_CODE, DST_CODE>
302	where
303	BitReader<E>: BitRead<E> + CodesRead<E>,
304	BitWriter<E>: BitWrite<E> + CodesWrite<E>,
305	{
306	}
307
308	impl<
309	E: Endianness,
310	D: BitDeserializer<E, BitReader<E>>,
311	const OUTDEGREE_CODE: usize,
312	const SRC_CODE: usize,
313	const DST_CODE: usize,
314	> Iterator for GroupedGapsIterator<E, D, OUTDEGREE_CODE, SRC_CODE, DST_CODE>
315	where
316	BitReader<E>: BitRead<E> + CodesRead<E>,
317	BitWriter<E>: BitWrite<E> + CodesWrite<E>,
318	{
319	type Item = ((usize, usize), D::DeserType);
320	fn next(&mut self) -> Option<Self::Item> {	67,915,955✔
321	if self.current >= self.len {	67,915,955✔
322	return None;	151✔
323	}
324	if self.dst_left == 0 {	67,915,804✔
325	// read a new source
326	let src_gap = ConstCode::<SRC_CODE>.read(&mut self.stream).ok()?;	20,344,365✔
327	self.src += src_gap as usize;	4,068,873✔
328	// read the outdegree
329	self.dst_left = ConstCode::<OUTDEGREE_CODE>.read(&mut self.stream).ok()? as usize;	16,275,492✔
330	self.prev_dst = 0;	4,068,873✔
331	}
332
333	let dst_gap = ConstCode::<DST_CODE>.read(&mut self.stream).ok()?;	339,579,020✔
334	let label = self.deserializer.deserialize(&mut self.stream).ok()?;	339,579,020✔
335	self.prev_dst += dst_gap as usize;	67,915,804✔
336	self.current += 1;	67,915,804✔
337	self.dst_left -= 1;	67,915,804✔
338	Some(((self.src, self.prev_dst), label))	67,915,804✔
339	}
340
NEW 341	fn size_hint(&self) -> (usize, Option<usize>) {	×
NEW 342	(self.len(), Some(self.len()))	×
343	}
344	}
345
346	impl<
347	E: Endianness,
348	D: BitDeserializer<E, BitReader<E>>,
349	const OUTDEGREE_CODE: usize,
350	const SRC_CODE: usize,
351	const DST_CODE: usize,
352	> ExactSizeIterator for GroupedGapsIterator<E, D, OUTDEGREE_CODE, SRC_CODE, DST_CODE>
353	where
354	BitReader<E>: BitRead<E> + CodesRead<E>,
355	BitWriter<E>: BitWrite<E> + CodesWrite<E>,
356	{
NEW 357	fn len(&self) -> usize {	×
NEW 358	self.len - self.current	×
359	}
360	}

vigna / webgraph-rs / 19278973738

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