terminus_store/layer/internal/
base.rs

1//! Base layer implementation.
2//!
3//! A base layer stores triple data without referring to a parent.
4use bytes::Bytes;
5use futures::stream::{Peekable, Stream, StreamExt};
6use futures::task::{Context, Poll};
7
8use super::super::builder::*;
9use super::super::id_map::*;
10use super::super::layer::*;
11use crate::layer::InternalLayer;
12use crate::{chrono_log, storage::*};
13use tdb_succinct::*;
14
15use std::io;
16use std::pin::Pin;
17
18/// A base layer.
19///
20/// This layer type has no parent, and therefore does not store any
21/// additions or removals. It stores all its triples plus indexes
22/// directly.
23#[derive(Clone)]
24pub struct BaseLayer {
25    pub(super) name: [u32; 5],
26    pub(super) node_dictionary: StringDict,
27    pub(super) predicate_dictionary: StringDict,
28    pub(super) value_dictionary: TypedDict,
29
30    pub(super) node_value_idmap: IdMap,
31    pub(super) predicate_idmap: IdMap,
32
33    pub(super) subjects: Option<MonotonicLogArray>,
34    pub(super) objects: Option<MonotonicLogArray>,
35
36    pub(super) s_p_adjacency_list: AdjacencyList,
37    pub(super) sp_o_adjacency_list: AdjacencyList,
38    pub(super) o_ps_adjacency_list: AdjacencyList,
39
40    pub(super) predicate_wavelet_tree: WaveletTree,
41}
42
43impl BaseLayer {
44    pub async fn load_from_files<F: FileLoad + FileStore>(
45        name: [u32; 5],
46        files: &BaseLayerFiles<F>,
47    ) -> io::Result<InternalLayer> {
48        let maps = files.map_all().await?;
49        Ok(Self::load(name, maps))
50    }
51
52    pub fn load(name: [u32; 5], maps: BaseLayerMaps) -> InternalLayer {
53        let node_dictionary = StringDict::parse(
54            maps.node_dictionary_maps.offsets_map,
55            maps.node_dictionary_maps.blocks_map,
56        );
57        let predicate_dictionary = StringDict::parse(
58            maps.predicate_dictionary_maps.offsets_map,
59            maps.predicate_dictionary_maps.blocks_map,
60        );
61        let value_dictionary = TypedDict::from_parts(
62            maps.value_dictionary_maps.types_present_map,
63            maps.value_dictionary_maps.type_offsets_map,
64            maps.value_dictionary_maps.offsets_map,
65            maps.value_dictionary_maps.blocks_map,
66        );
67
68        let node_value_idmap = match maps.id_map_maps.node_value_idmap_maps {
69            None => IdMap::default(),
70            Some(maps) => IdMap::from_maps(
71                maps,
72                util::calculate_width(
73                    (node_dictionary.num_entries() + value_dictionary.num_entries()) as u64,
74                ),
75            ),
76        };
77
78        let predicate_idmap = match maps.id_map_maps.predicate_idmap_maps {
79            None => IdMap::default(),
80            Some(map) => IdMap::from_maps(
81                map,
82                util::calculate_width(predicate_dictionary.num_entries() as u64),
83            ),
84        };
85
86        let subjects = maps.subjects_map.map(|subjects_map| {
87            MonotonicLogArray::from_logarray(LogArray::parse(subjects_map).unwrap())
88        });
89        let objects = maps.objects_map.map(|objects_map| {
90            MonotonicLogArray::from_logarray(LogArray::parse(objects_map).unwrap())
91        });
92
93        let s_p_adjacency_list = AdjacencyList::parse(
94            maps.s_p_adjacency_list_maps.nums_map,
95            maps.s_p_adjacency_list_maps.bitindex_maps.bits_map,
96            maps.s_p_adjacency_list_maps.bitindex_maps.blocks_map,
97            maps.s_p_adjacency_list_maps.bitindex_maps.sblocks_map,
98        );
99        let sp_o_adjacency_list = AdjacencyList::parse(
100            maps.sp_o_adjacency_list_maps.nums_map,
101            maps.sp_o_adjacency_list_maps.bitindex_maps.bits_map,
102            maps.sp_o_adjacency_list_maps.bitindex_maps.blocks_map,
103            maps.sp_o_adjacency_list_maps.bitindex_maps.sblocks_map,
104        );
105        let o_ps_adjacency_list = AdjacencyList::parse(
106            maps.o_ps_adjacency_list_maps.nums_map,
107            maps.o_ps_adjacency_list_maps.bitindex_maps.bits_map,
108            maps.o_ps_adjacency_list_maps.bitindex_maps.blocks_map,
109            maps.o_ps_adjacency_list_maps.bitindex_maps.sblocks_map,
110        );
111
112        let predicate_wavelet_tree_width = s_p_adjacency_list.nums().width();
113        let predicate_wavelet_tree = WaveletTree::from_parts(
114            BitIndex::from_maps(
115                maps.predicate_wavelet_tree_maps.bits_map,
116                maps.predicate_wavelet_tree_maps.blocks_map,
117                maps.predicate_wavelet_tree_maps.sblocks_map,
118            ),
119            predicate_wavelet_tree_width,
120        );
121
122        InternalLayer::Base(BaseLayer {
123            name,
124            node_dictionary,
125            predicate_dictionary,
126            value_dictionary,
127
128            node_value_idmap,
129            predicate_idmap,
130
131            subjects,
132            objects,
133
134            s_p_adjacency_list,
135            sp_o_adjacency_list,
136
137            o_ps_adjacency_list,
138
139            predicate_wavelet_tree,
140        })
141    }
142}
143
144/// A builder for a base layer.
145///
146/// This builder takes node, predicate and value strings in lexical
147/// order through the corresponding `add_<thing>` methods. When
148/// they're all added, `into_phase2()` is to be called to turn this
149/// builder into a second builder that takes triple data.
150pub struct BaseLayerFileBuilder<F: 'static + FileLoad + FileStore> {
151    files: BaseLayerFiles<F>,
152
153    builder: DictionarySetFileBuilder<F>,
154}
155
156impl<F: 'static + FileLoad + FileStore + Clone> BaseLayerFileBuilder<F> {
157    /// Create the builder from the given files.
158    pub async fn from_files(files: &BaseLayerFiles<F>) -> io::Result<Self> {
159        let builder = DictionarySetFileBuilder::from_files(
160            files.node_dictionary_files.clone(),
161            files.predicate_dictionary_files.clone(),
162            files.value_dictionary_files.clone(),
163        )
164        .await?;
165
166        Ok(BaseLayerFileBuilder {
167            files: files.clone(),
168            builder,
169        })
170    }
171
172    /// Add a node string.
173    ///
174    /// Panics if the given node string is not a lexical successor of the previous node string.
175    pub fn add_node(&mut self, node: &str) -> u64 {
176        let id = self.builder.add_node(node);
177
178        id
179    }
180
181    pub fn add_node_bytes(&mut self, node: Bytes) -> u64 {
182        self.builder.add_node_bytes(node)
183    }
184
185    /// Add a predicate string.
186    ///
187    /// Panics if the given predicate string is not a lexical successor of the previous node string.
188    pub fn add_predicate(&mut self, predicate: &str) -> u64 {
189        let id = self.builder.add_predicate(predicate);
190
191        id
192    }
193
194    pub fn add_predicate_bytes(&mut self, predicate: Bytes) -> u64 {
195        self.builder.add_predicate_bytes(predicate)
196    }
197
198    /// Add a value string.
199    ///
200    /// Panics if the given value string is not a lexical successor of the previous value string.
201    pub fn add_value(&mut self, value: TypedDictEntry) -> u64 {
202        let id = self.builder.add_value(value);
203
204        id
205    }
206
207    /// Add nodes from an iterable.
208    ///
209    /// Panics if the nodes are not in lexical order, or if previous added nodes are a lexical succesor of any of these nodes.
210    pub fn add_nodes<I: 'static + IntoIterator<Item = String> + Send>(
211        &mut self,
212        nodes: I,
213    ) -> Vec<u64>
214    where
215        <I as std::iter::IntoIterator>::IntoIter: Unpin + Send + Sync,
216        I: Unpin + Sync,
217    {
218        let ids = self.builder.add_nodes(nodes);
219
220        ids
221    }
222
223    pub fn add_nodes_bytes<I: 'static + IntoIterator<Item = Bytes> + Send>(
224        &mut self,
225        nodes: I,
226    ) -> Vec<u64>
227    where
228        <I as std::iter::IntoIterator>::IntoIter: Unpin + Send + Sync,
229        I: Unpin + Sync,
230    {
231        let ids = self.builder.add_nodes_bytes(nodes);
232
233        ids
234    }
235
236    /// Add predicates from an iterable.
237    ///
238    /// Panics if the predicates are not in lexical order, or if previous added predicates are a lexical succesor of any of these predicates.
239    pub fn add_predicates<I: 'static + IntoIterator<Item = String> + Send>(
240        &mut self,
241        predicates: I,
242    ) -> Vec<u64>
243    where
244        <I as std::iter::IntoIterator>::IntoIter: Unpin + Send + Sync,
245        I: Unpin + Sync,
246    {
247        let ids = self.builder.add_predicates(predicates);
248
249        ids
250    }
251
252    pub fn add_predicates_bytes<I: 'static + IntoIterator<Item = Bytes> + Send>(
253        &mut self,
254        predicates: I,
255    ) -> Vec<u64>
256    where
257        <I as std::iter::IntoIterator>::IntoIter: Unpin + Send + Sync,
258        I: Unpin + Sync,
259    {
260        let ids = self.builder.add_predicates_bytes(predicates);
261
262        ids
263    }
264
265    /// Add values from an iterable.
266    ///
267    /// Panics if the values are not in lexical order, or if previous added values are a lexical succesor of any of these values.
268    pub fn add_values<I: 'static + IntoIterator<Item = TypedDictEntry> + Send>(
269        &mut self,
270        values: I,
271    ) -> Vec<u64>
272    where
273        <I as std::iter::IntoIterator>::IntoIter: Unpin + Send + Sync,
274        I: Unpin + Sync,
275    {
276        let ids = self.builder.add_values(values);
277
278        ids
279    }
280
281    /// Turn this builder into a phase 2 builder that will take triple data.
282    pub async fn into_phase2(self) -> io::Result<BaseLayerFileBuilderPhase2<F>> {
283        let BaseLayerFileBuilder { files, builder } = self;
284
285        builder.finalize().await?;
286
287        let node_dict_blocks_map = files.node_dictionary_files.blocks_file.map().await?;
288        let node_dict_offsets_map = files.node_dictionary_files.offsets_file.map().await?;
289        let predicate_dict_blocks_map = files.predicate_dictionary_files.blocks_file.map().await?;
290        let predicate_dict_offsets_map =
291            files.predicate_dictionary_files.offsets_file.map().await?;
292        let value_dict_types_present_map = files
293            .value_dictionary_files
294            .types_present_file
295            .map()
296            .await?;
297        let value_dict_type_offsets_map =
298            files.value_dictionary_files.type_offsets_file.map().await?;
299        let value_dict_blocks_map = files.value_dictionary_files.blocks_file.map().await?;
300        let value_dict_offsets_map = files.value_dictionary_files.offsets_file.map().await?;
301
302        let node_dict = StringDict::parse(node_dict_offsets_map, node_dict_blocks_map);
303        let pred_dict = StringDict::parse(predicate_dict_offsets_map, predicate_dict_blocks_map);
304        let val_dict = TypedDict::from_parts(
305            value_dict_types_present_map,
306            value_dict_type_offsets_map,
307            value_dict_offsets_map,
308            value_dict_blocks_map,
309        );
310
311        // TODO: it is a bit silly to parse the dictionaries just for this. surely we can get the counts in an easier way?
312        let num_nodes = node_dict.num_entries();
313        let num_predicates = pred_dict.num_entries();
314        let num_values = val_dict.num_entries();
315
316        BaseLayerFileBuilderPhase2::new(files, num_nodes, num_predicates, num_values).await
317    }
318}
319
320/// Second phase of base layer building.
321///
322/// This builder takes ordered triple data. When all data has been
323/// added, `finalize()` will build a layer.
324pub struct BaseLayerFileBuilderPhase2<F: 'static + FileLoad + FileStore> {
325    files: BaseLayerFiles<F>,
326
327    builder: TripleFileBuilder<F>,
328}
329
330impl<F: 'static + FileLoad + FileStore> BaseLayerFileBuilderPhase2<F> {
331    pub async fn new(
332        files: BaseLayerFiles<F>,
333
334        num_nodes: usize,
335        num_predicates: usize,
336        num_values: usize,
337    ) -> io::Result<Self> {
338        let builder = TripleFileBuilder::new(
339            files.s_p_adjacency_list_files.clone(),
340            files.sp_o_adjacency_list_files.clone(),
341            num_nodes,
342            num_predicates,
343            num_values,
344            None,
345        )
346        .await?;
347
348        Ok(BaseLayerFileBuilderPhase2 { files, builder })
349    }
350
351    /// Add the given subject, predicate and object.
352    ///
353    /// This will panic if a greater triple has already been added.
354    pub async fn add_triple(
355        &mut self,
356        subject: u64,
357        predicate: u64,
358        object: u64,
359    ) -> io::Result<()> {
360        self.builder.add_triple(subject, predicate, object).await
361    }
362
363    /// Add the given triple.
364    ///
365    /// This will panic if a greater triple has already been added.
366    pub async fn add_id_triples<I: 'static + IntoIterator<Item = IdTriple>>(
367        &mut self,
368        triples: I,
369    ) -> io::Result<()>
370    where
371        <I as std::iter::IntoIterator>::IntoIter: Unpin + Send,
372    {
373        self.builder.add_id_triples(triples).await
374    }
375
376    pub(crate) async fn partial_finalize(self) -> io::Result<BaseLayerFiles<F>> {
377        self.builder.finalize().await?;
378        chrono_log!("finalized base triples builder");
379
380        Ok(self.files)
381    }
382
383    pub async fn finalize(self) -> io::Result<()> {
384        self.builder.finalize().await?;
385        chrono_log!("finalized base triples builder");
386        let s_p_adjacency_list_files = self.files.s_p_adjacency_list_files.clone();
387        let sp_o_adjacency_list_files = self.files.sp_o_adjacency_list_files.clone();
388        let o_ps_adjacency_list_files = self.files.o_ps_adjacency_list_files.clone();
389        let predicate_wavelet_tree_files = self.files.predicate_wavelet_tree_files.clone();
390        build_indexes(
391            s_p_adjacency_list_files,
392            sp_o_adjacency_list_files,
393            o_ps_adjacency_list_files,
394            None,
395            predicate_wavelet_tree_files,
396        )
397        .await?;
398
399        chrono_log!("finalized base builder");
400
401        Ok(())
402    }
403}
404
405pub struct BaseTripleStream<S: Stream<Item = io::Result<(u64, u64)>> + Send> {
406    s_p_stream: Peekable<S>,
407    sp_o_stream: Peekable<S>,
408    last_s_p: (u64, u64),
409    last_sp: u64,
410}
411
412impl<S: Stream<Item = io::Result<(u64, u64)>> + Unpin + Send> BaseTripleStream<S> {
413    pub fn new(s_p_stream: S, sp_o_stream: S) -> BaseTripleStream<S> {
414        BaseTripleStream {
415            s_p_stream: s_p_stream.peekable(),
416            sp_o_stream: sp_o_stream.peekable(),
417            last_s_p: (0, 0),
418            last_sp: 0,
419        }
420    }
421}
422
423impl<S: Stream<Item = io::Result<(u64, u64)>> + Unpin + Send> Stream for BaseTripleStream<S> {
424    type Item = io::Result<(u64, u64, u64)>;
425
426    fn poll_next(
427        mut self: Pin<&mut Self>,
428        cx: &mut Context,
429    ) -> Poll<Option<io::Result<(u64, u64, u64)>>> {
430        let peeked = Pin::new(&mut self.sp_o_stream).poll_peek(cx);
431        match peeked {
432            Poll::Ready(Some(Ok((sp, o)))) => {
433                let sp = *sp;
434                let o = *o;
435                if sp > self.last_sp {
436                    let peeked = Pin::new(&mut self.s_p_stream).poll_peek(cx);
437                    match peeked {
438                        Poll::Ready(None) => Poll::Ready(Some(Err(io::Error::new(
439                            io::ErrorKind::UnexpectedEof,
440                            "unexpected end of s_p_stream",
441                        )))),
442                        Poll::Ready(Some(Ok((s, p)))) => {
443                            let s = *s;
444                            let p = *p;
445                            util::assert_poll_next(Pin::new(&mut self.s_p_stream), cx).unwrap();
446                            util::assert_poll_next(Pin::new(&mut self.sp_o_stream), cx).unwrap();
447
448                            self.last_s_p = (s, p);
449                            self.last_sp = sp;
450
451                            Poll::Ready(Some(Ok((s, p, o))))
452                        }
453                        Poll::Ready(Some(Err(_))) => Poll::Ready(Some(Err(
454                            util::assert_poll_next(Pin::new(&mut self.s_p_stream), cx)
455                                .err()
456                                .unwrap(),
457                        ))),
458                        Poll::Pending => Poll::Pending,
459                    }
460                } else {
461                    util::assert_poll_next(Pin::new(&mut self.sp_o_stream), cx).unwrap();
462
463                    Poll::Ready(Some(Ok((self.last_s_p.0, self.last_s_p.1, o))))
464                }
465            }
466            Poll::Ready(Some(Err(_))) => Poll::Ready(Some(Err(util::assert_poll_next(
467                Pin::new(&mut self.sp_o_stream),
468                cx,
469            )
470            .err()
471            .unwrap()))),
472            Poll::Ready(None) => Poll::Ready(None),
473            Poll::Pending => Poll::Pending,
474        }
475    }
476}
477
478pub async fn open_base_triple_stream<F: 'static + FileLoad + FileStore>(
479    s_p_files: AdjacencyListFiles<F>,
480    sp_o_files: AdjacencyListFiles<F>,
481) -> io::Result<impl Stream<Item = io::Result<(u64, u64, u64)>> + Unpin + Send> {
482    let s_p_stream =
483        adjacency_list_stream_pairs(s_p_files.bitindex_files.bits_file, s_p_files.nums_file)
484            .await?;
485    let sp_o_stream =
486        adjacency_list_stream_pairs(sp_o_files.bitindex_files.bits_file, sp_o_files.nums_file)
487            .await?;
488
489    Ok(BaseTripleStream::new(s_p_stream, sp_o_stream))
490}
491
492#[cfg(test)]
493pub mod base_tests {
494    use super::*;
495    use crate::storage::memory::*;
496    use futures::stream::TryStreamExt;
497
498    pub fn base_layer_files() -> BaseLayerFiles<MemoryBackedStore> {
499        // TODO inline
500        base_layer_memory_files()
501    }
502
503    pub async fn example_base_layer_files() -> io::Result<BaseLayerFiles<MemoryBackedStore>> {
504        let nodes = vec!["aaaaa", "baa", "bbbbb", "ccccc", "mooo"];
505        let predicates = vec!["abcde", "fghij", "klmno", "lll"];
506        let values = vec!["chicken", "cow", "dog", "pig", "zebra"];
507
508        let base_layer_files = base_layer_files();
509
510        let mut builder = BaseLayerFileBuilder::from_files(&base_layer_files).await?;
511
512        builder.add_nodes(nodes.into_iter().map(|s| s.to_string()));
513        builder.add_predicates(predicates.into_iter().map(|s| s.to_string()));
514        builder.add_values(values.into_iter().map(|s| String::make_entry(&s)));
515
516        let mut builder = builder.into_phase2().await?;
517
518        builder.add_triple(1, 1, 1).await?;
519        builder.add_triple(2, 1, 1).await?;
520        builder.add_triple(2, 1, 3).await?;
521        builder.add_triple(2, 3, 6).await?;
522        builder.add_triple(3, 2, 5).await?;
523        builder.add_triple(3, 3, 6).await?;
524        builder.add_triple(4, 3, 6).await?;
525
526        builder.finalize().await?;
527
528        Ok(base_layer_files)
529    }
530
531    pub async fn example_base_layer() -> InternalLayer {
532        let base_layer_files = example_base_layer_files().await.unwrap();
533
534        BaseLayer::load_from_files([1, 2, 3, 4, 5], &base_layer_files)
535            .await
536            .unwrap()
537    }
538
539    #[tokio::test]
540    async fn build_and_query_base_layer() {
541        let layer = example_base_layer().await;
542
543        assert!(layer.triple_exists(1, 1, 1));
544        assert!(layer.triple_exists(2, 1, 1));
545        assert!(layer.triple_exists(2, 1, 3));
546        assert!(layer.triple_exists(2, 3, 6));
547        assert!(layer.triple_exists(3, 2, 5));
548        assert!(layer.triple_exists(3, 3, 6));
549        assert!(layer.triple_exists(4, 3, 6));
550
551        assert!(!layer.triple_exists(2, 2, 0));
552    }
553
554    #[tokio::test]
555    async fn dictionary_entries_in_base() {
556        let base_layer = example_base_layer().await;
557
558        assert_eq!(3, base_layer.subject_id("bbbbb").unwrap());
559        assert_eq!(2, base_layer.predicate_id("fghij").unwrap());
560        assert_eq!(1, base_layer.object_node_id("aaaaa").unwrap());
561        assert_eq!(
562            6,
563            base_layer
564                .object_value_id(&String::make_entry(&"chicken"))
565                .unwrap()
566        );
567
568        assert_eq!("bbbbb", base_layer.id_subject(3).unwrap());
569        assert_eq!("fghij", base_layer.id_predicate(2).unwrap());
570        assert_eq!(
571            ObjectType::Node("aaaaa".to_string()),
572            base_layer.id_object(1).unwrap()
573        );
574        assert_eq!(
575            ObjectType::Value(String::make_entry(&"chicken")),
576            base_layer.id_object(6).unwrap()
577        );
578    }
579
580    #[tokio::test]
581    async fn everything_iterator() {
582        let layer = example_base_layer().await;
583        let triples: Vec<_> = layer
584            .triples()
585            .map(|t| (t.subject, t.predicate, t.object))
586            .collect();
587
588        assert_eq!(
589            vec![
590                (1, 1, 1),
591                (2, 1, 1),
592                (2, 1, 3),
593                (2, 3, 6),
594                (3, 2, 5),
595                (3, 3, 6),
596                (4, 3, 6)
597            ],
598            triples
599        );
600    }
601
602    #[tokio::test]
603    async fn lookup_by_object() {
604        let layer = example_base_layer().await;
605
606        let triples: Vec<_> = layer
607            .triples_o(1)
608            .map(|t| (t.subject, t.predicate, t.object))
609            .collect();
610        assert_eq!(vec![(1, 1, 1), (2, 1, 1)], triples);
611
612        let triples: Vec<_> = layer
613            .triples_o(3)
614            .map(|t| (t.subject, t.predicate, t.object))
615            .collect();
616        assert_eq!(vec![(2, 1, 3)], triples);
617
618        let triples: Vec<_> = layer
619            .triples_o(5)
620            .map(|t| (t.subject, t.predicate, t.object))
621            .collect();
622        assert_eq!(vec![(3, 2, 5)], triples);
623
624        let triples: Vec<_> = layer
625            .triples_o(6)
626            .map(|t| (t.subject, t.predicate, t.object))
627            .collect();
628        assert_eq!(vec![(2, 3, 6), (3, 3, 6), (4, 3, 6)], triples);
629    }
630
631    #[tokio::test]
632    async fn lookup_by_predicate() {
633        let layer = example_base_layer().await;
634
635        let pairs: Vec<_> = layer
636            .triples_p(1)
637            .map(|t| (t.subject, t.predicate, t.object))
638            .collect();
639
640        assert_eq!(vec![(1, 1, 1), (2, 1, 1), (2, 1, 3)], pairs);
641
642        let pairs: Vec<_> = layer
643            .triples_p(2)
644            .map(|t| (t.subject, t.predicate, t.object))
645            .collect();
646
647        assert_eq!(vec![(3, 2, 5)], pairs);
648
649        let pairs: Vec<_> = layer
650            .triples_p(3)
651            .map(|t| (t.subject, t.predicate, t.object))
652            .collect();
653
654        assert_eq!(vec![(2, 3, 6), (3, 3, 6), (4, 3, 6)], pairs);
655
656        assert!(layer.triples_p(4).next().is_none());
657    }
658
659    #[tokio::test]
660    async fn create_empty_base_layer() {
661        let base_layer_files = base_layer_files();
662        let builder = BaseLayerFileBuilder::from_files(&base_layer_files)
663            .await
664            .unwrap();
665
666        let builder = builder.into_phase2().await.unwrap();
667        builder.finalize().await.unwrap();
668
669        let layer = BaseLayer::load_from_files([1, 2, 3, 4, 5], &base_layer_files)
670            .await
671            .unwrap();
672
673        assert_eq!(0, layer.node_and_value_count());
674        assert_eq!(0, layer.predicate_count());
675    }
676
677    #[tokio::test]
678    async fn stream_base_triples() {
679        let layer_files = example_base_layer_files().await.unwrap();
680
681        let stream = open_base_triple_stream(
682            layer_files.s_p_adjacency_list_files,
683            layer_files.sp_o_adjacency_list_files,
684        )
685        .await
686        .unwrap();
687
688        let triples: Vec<_> = stream.try_collect().await.unwrap();
689
690        assert_eq!(
691            vec![
692                (1, 1, 1),
693                (2, 1, 1),
694                (2, 1, 3),
695                (2, 3, 6),
696                (3, 2, 5),
697                (3, 3, 6),
698                (4, 3, 6)
699            ],
700            triples
701        );
702    }
703
704    #[tokio::test]
705    async fn count_triples() {
706        let layer = example_base_layer().await;
707
708        assert_eq!(7, layer.internal_triple_layer_addition_count());
709        assert_eq!(0, layer.internal_triple_layer_removal_count());
710        assert_eq!(7, layer.triple_addition_count());
711        assert_eq!(0, layer.triple_removal_count());
712        assert_eq!(7, layer.triple_count());
713    }
714
715    #[tokio::test]
716    async fn count_triples_of_empty_base_layer() {
717        let layer_files = base_layer_files();
718        let builder = BaseLayerFileBuilder::from_files(&layer_files)
719            .await
720            .unwrap();
721        builder
722            .into_phase2()
723            .await
724            .unwrap()
725            .finalize()
726            .await
727            .unwrap();
728        let layer = BaseLayer::load_from_files([1, 2, 3, 4, 5], &layer_files)
729            .await
730            .unwrap();
731
732        assert_eq!(0, layer.internal_triple_layer_addition_count());
733        assert_eq!(0, layer.internal_triple_layer_removal_count());
734        assert_eq!(0, layer.triple_count());
735        assert_eq!(0, layer.triple_addition_count());
736        assert_eq!(0, layer.triple_removal_count());
737    }
738}
terminus_store/layer/internal/base.rs

terminus_store/layer/internal/
base.rs
