Skip to main content

terminus_store/store/
mod.rs

1//! High-level API for working with terminus-store.
2//!
3//! It is expected that most users of this library will work exclusively with the types contained in this module.
4pub mod sync;
5
6use std::path::{Path, PathBuf};
7use std::sync::{Arc, RwLock};
8
9use crate::layer::{IdTriple, Layer, LayerBuilder, LayerCounts, ObjectType, ValueTriple};
10use crate::storage::archive::{ArchiveLayerStore, DirectoryArchiveBackend, LruArchiveBackend};
11use crate::storage::directory::{DirectoryLabelStore, DirectoryLayerStore};
12use crate::storage::memory::{MemoryLabelStore, MemoryLayerStore};
13use crate::storage::{CachedLayerStore, LabelStore, LayerStore, LockingHashMapLayerCache};
14use tdb_succinct::TypedDictEntry;
15
16use std::io;
17
18use async_trait::async_trait;
19use rayon::prelude::*;
20
21/// A store, storing a set of layers and database labels pointing to these layers.
22#[derive(Clone)]
23pub struct Store {
24    label_store: Arc<dyn LabelStore>,
25    layer_store: Arc<dyn LayerStore>,
26}
27
28/// A wrapper over a SimpleLayerBuilder, providing a thread-safe sharable interface.
29///
30/// The SimpleLayerBuilder requires one to have a mutable reference to
31/// the underlying LayerBuilder, and on commit it will be
32/// consumed. This builder only requires an immutable reference, and
33/// uses a futures-aware read-write lock to synchronize access to it
34/// between threads. Also, rather than consuming itself on commit,
35/// this wrapper will simply mark itself as having committed,
36/// returning errors on further calls.
37#[derive(Clone)]
38pub struct StoreLayerBuilder {
39    parent: Option<Arc<dyn Layer>>,
40    builder: Arc<RwLock<Option<Box<dyn LayerBuilder>>>>,
41    name: [u32; 5],
42    store: Store,
43}
44
45impl StoreLayerBuilder {
46    async fn new(store: Store) -> io::Result<Self> {
47        let builder = store.layer_store.create_base_layer().await?;
48
49        Ok(Self {
50            parent: builder.parent(),
51            name: builder.name(),
52            builder: Arc::new(RwLock::new(Some(builder))),
53            store,
54        })
55    }
56
57    fn wrap(builder: Box<dyn LayerBuilder>, store: Store) -> Self {
58        StoreLayerBuilder {
59            parent: builder.parent(),
60            name: builder.name(),
61            builder: Arc::new(RwLock::new(Some(builder))),
62            store,
63        }
64    }
65
66    pub fn with_builder<R, F: FnOnce(&mut Box<dyn LayerBuilder>) -> R>(
67        &self,
68        f: F,
69    ) -> Result<R, io::Error> {
70        let mut builder = self
71            .builder
72            .write()
73            .expect("rwlock write should always succeed");
74        match (*builder).as_mut() {
75            None => Err(io::Error::new(
76                io::ErrorKind::InvalidData,
77                "builder has already been committed",
78            )),
79            Some(builder) => Ok(f(builder)),
80        }
81    }
82
83    /// Returns the name of the layer being built.
84    pub fn name(&self) -> [u32; 5] {
85        self.name
86    }
87
88    /// Returns the parent layer this builder is building on top of, if any.
89    ///
90    /// If there's no parent, this returns None.
91    pub fn parent(&self) -> Option<Arc<dyn Layer>> {
92        self.parent.clone()
93    }
94
95    /// Add a string triple.
96    pub fn add_value_triple(&self, triple: ValueTriple) -> Result<(), io::Error> {
97        self.with_builder(move |b| b.add_value_triple(triple))
98    }
99
100    /// Add an id triple.
101    pub fn add_id_triple(&self, triple: IdTriple) -> Result<(), io::Error> {
102        self.with_builder(move |b| b.add_id_triple(triple))
103    }
104
105    /// Remove a string triple.
106    pub fn remove_value_triple(&self, triple: ValueTriple) -> Result<(), io::Error> {
107        self.with_builder(move |b| b.remove_value_triple(triple))
108    }
109
110    /// Remove an id triple.
111    pub fn remove_id_triple(&self, triple: IdTriple) -> Result<(), io::Error> {
112        self.with_builder(move |b| b.remove_id_triple(triple))
113    }
114
115    /// Returns true if this layer has been committed, and false otherwise.
116    pub fn committed(&self) -> bool {
117        self.builder
118            .read()
119            .expect("rwlock write should always succeed")
120            .is_none()
121    }
122
123    /// Commit the layer to storage without loading the resulting layer.
124    pub async fn commit_no_load(&self) -> io::Result<()> {
125        let mut builder = None;
126        {
127            let mut guard = self
128                .builder
129                .write()
130                .expect("rwlock write should always succeed");
131
132            // Setting the builder to None ensures that committed() detects we already committed (or tried to do so anyway)
133            std::mem::swap(&mut builder, &mut guard);
134        }
135
136        match builder {
137            None => {
138                return Err(io::Error::new(
139                    io::ErrorKind::InvalidData,
140                    "builder has already been committed",
141                ))
142            }
143            Some(builder) => {
144                let id = builder.name();
145                builder.commit_boxed().await?;
146                self.store.layer_store.finalize_layer(id).await
147            }
148        }
149    }
150
151    /// Commit the layer to storage.
152    pub async fn commit(&self) -> io::Result<StoreLayer> {
153        let name = self.name;
154        self.commit_no_load().await?;
155
156        let layer = self.store.layer_store.get_layer(name).await?;
157        Ok(StoreLayer::wrap(
158            layer.expect("layer that was just created was not found in store"),
159            self.store.clone(),
160        ))
161    }
162
163    /// Apply all triples added and removed by a layer to this builder.
164    ///
165    /// This is a way to 'cherry-pick' a layer on top of another
166    /// layer, without caring about its history.
167    pub async fn apply_delta(&self, delta: &StoreLayer) -> Result<(), io::Error> {
168        // create a child builder and use it directly
169        // first check what dictionary entries we don't know about, add those
170        let triple_additions = delta.triple_additions().await?;
171        let triple_removals = delta.triple_removals().await?;
172        rayon::join(
173            move || {
174                triple_additions.par_bridge().for_each(|t| {
175                    delta
176                        .id_triple_to_string(&t)
177                        .map(|st| self.add_value_triple(st));
178                });
179            },
180            move || {
181                triple_removals.par_bridge().for_each(|t| {
182                    delta
183                        .id_triple_to_string(&t)
184                        .map(|st| self.remove_value_triple(st));
185                })
186            },
187        );
188
189        Ok(())
190    }
191
192    /// Apply the changes required to change our parent layer into the given layer.
193    pub fn apply_diff(&self, other: &StoreLayer) -> Result<(), io::Error> {
194        // create a child builder and use it directly
195        // first check what dictionary entries we don't know about, add those
196        rayon::join(
197            || {
198                if let Some(this) = self.parent() {
199                    this.triples().par_bridge().for_each(|t| {
200                        if let Some(st) = this.id_triple_to_string(&t) {
201                            if !other.value_triple_exists(&st) {
202                                self.remove_value_triple(st).unwrap()
203                            }
204                        }
205                    })
206                };
207            },
208            || {
209                other.triples().par_bridge().for_each(|t| {
210                    if let Some(st) = other.id_triple_to_string(&t) {
211                        if let Some(this) = self.parent() {
212                            if !this.value_triple_exists(&st) {
213                                self.add_value_triple(st).unwrap()
214                            }
215                        } else {
216                            self.add_value_triple(st).unwrap()
217                        };
218                    }
219                })
220            },
221        );
222
223        Ok(())
224    }
225}
226
227/// A layer that keeps track of the store it came out of, allowing the creation of a layer builder on top of this layer.
228///
229/// This type of layer supports querying what was added and what was
230/// removed in this layer. This can not be done in general, because
231/// the layer that has been loaded may not be the layer that was
232/// originally built. This happens whenever a rollup is done. A rollup
233/// will create a new layer that bundles the changes of various
234/// layers. It allows for more efficient querying, but loses the
235/// ability to do these delta queries directly. In order to support
236/// them anyway, the StoreLayer will dynamically load in the relevant
237/// files to perform the requested addition or removal query method.
238#[derive(Clone)]
239pub struct StoreLayer {
240    // TODO this Arc here is not great
241    layer: Arc<dyn Layer>,
242    store: Store,
243}
244
245impl StoreLayer {
246    fn wrap(layer: Arc<dyn Layer>, store: Store) -> Self {
247        StoreLayer { layer, store }
248    }
249
250    /// Create a layer builder based on this layer.
251    pub async fn open_write(&self) -> io::Result<StoreLayerBuilder> {
252        let layer = self
253            .store
254            .layer_store
255            .create_child_layer(self.layer.name())
256            .await?;
257
258        Ok(StoreLayerBuilder::wrap(layer, self.store.clone()))
259    }
260
261    /// Returns the parent of this layer, if any, or None if this layer has no parent.
262    pub async fn parent(&self) -> io::Result<Option<StoreLayer>> {
263        let parent_name = self.layer.parent_name();
264
265        match parent_name {
266            None => Ok(None),
267            Some(parent_name) => match self.store.layer_store.get_layer(parent_name).await? {
268                None => Err(io::Error::new(
269                    io::ErrorKind::NotFound,
270                    "parent layer not found even though it should exist",
271                )),
272                Some(layer) => Ok(Some(StoreLayer::wrap(layer, self.store.clone()))),
273            },
274        }
275    }
276
277    pub async fn squash_upto(&self, upto: &StoreLayer) -> io::Result<StoreLayer> {
278        let layer_opt = self.store.layer_store.get_layer(self.name()).await?;
279        let layer =
280            layer_opt.ok_or_else(|| io::Error::new(io::ErrorKind::NotFound, "layer not found"))?;
281        let name = self
282            .store
283            .layer_store
284            .squash_upto(layer, upto.name())
285            .await?;
286        Ok(self
287            .store
288            .get_layer_from_id(name)
289            .await?
290            .expect("layer that was just created doesn't exist"))
291    }
292
293    /// Create a new base layer consisting of all triples in this layer, as well as all its ancestors.
294    ///
295    /// It is a good idea to keep layer stacks small, meaning, to only
296    /// have a handful of ancestors for a layer. The more layers there
297    /// are, the longer queries take. Squash is one approach of
298    /// accomplishing this. Rollup is another. Squash is the better
299    /// option if you do not care for history, as it throws away all
300    /// data that you no longer need.
301    pub async fn squash(&self) -> io::Result<StoreLayer> {
302        let layer_opt = self.store.layer_store.get_layer(self.name()).await?;
303        let layer =
304            layer_opt.ok_or_else(|| io::Error::new(io::ErrorKind::NotFound, "layer not found"))?;
305        let name = self.store.layer_store.squash(layer).await?;
306        Ok(self
307            .store
308            .get_layer_from_id(name)
309            .await?
310            .expect("layer that was just created doesn't exist"))
311    }
312
313    /// Create a new rollup layer which rolls up all triples in this layer, as well as all its ancestors.
314    ///
315    /// It is a good idea to keep layer stacks small, meaning, to only
316    /// have a handful of ancestors for a layer. The more layers there
317    /// are, the longer queries take. Rollup is one approach of
318    /// accomplishing this. Squash is another. Rollup is the better
319    /// option if you need to retain history.
320    pub async fn rollup(&self) -> io::Result<()> {
321        let store1 = self.store.layer_store.clone();
322        // TODO: This is awkward, we should have a way to get the internal layer
323        let layer_opt = store1.get_layer(self.name()).await?;
324        let layer =
325            layer_opt.ok_or_else(|| io::Error::new(io::ErrorKind::NotFound, "layer not found"))?;
326        let store2 = self.store.layer_store.clone();
327        store2.rollup(layer).await?;
328        Ok(())
329    }
330
331    /// Create a new rollup layer which rolls up all triples in this layer, as well as all ancestors up to (but not including) the given ancestor.
332    ///
333    /// It is a good idea to keep layer stacks small, meaning, to only
334    /// have a handful of ancestors for a layer. The more layers there
335    /// are, the longer queries take. Rollup is one approach of
336    /// accomplishing this. Squash is another. Rollup is the better
337    /// option if you need to retain history.
338    pub async fn rollup_upto(&self, upto: &StoreLayer) -> io::Result<()> {
339        let store1 = self.store.layer_store.clone();
340        // TODO: This is awkward, we should have a way to get the internal layer
341        let layer_opt = store1.get_layer(self.name()).await?;
342        let layer =
343            layer_opt.ok_or_else(|| io::Error::new(io::ErrorKind::NotFound, "label not found"))?;
344        let store2 = self.store.layer_store.clone();
345        store2.rollup_upto(layer, upto.name()).await?;
346        Ok(())
347    }
348
349    /// Like rollup_upto, rolls up upto the given layer. However, if
350    /// this layer is a rollup layer, this will roll up upto that
351    /// rollup.
352    pub async fn imprecise_rollup_upto(&self, upto: &StoreLayer) -> io::Result<()> {
353        let store1 = self.store.layer_store.clone();
354        // TODO: This is awkward, we should have a way to get the internal layer
355        let layer_opt = store1.get_layer(self.name()).await?;
356        let layer =
357            layer_opt.ok_or_else(|| io::Error::new(io::ErrorKind::NotFound, "label not found"))?;
358        let store2 = self.store.layer_store.clone();
359        store2.imprecise_rollup_upto(layer, upto.name()).await?;
360        Ok(())
361    }
362
363    /// Returns a future that yields true if this triple has been added in this layer, or false if it doesn't.
364    ///
365    /// Since this operation will involve io when this layer is a
366    /// rollup layer, io errors may occur.
367    pub async fn triple_addition_exists(
368        &self,
369        subject: u64,
370        predicate: u64,
371        object: u64,
372    ) -> io::Result<bool> {
373        self.store
374            .layer_store
375            .triple_addition_exists(self.layer.name(), subject, predicate, object)
376            .await
377    }
378
379    /// Returns a future that yields true if this triple has been removed in this layer, or false if it doesn't.
380    ///
381    /// Since this operation will involve io when this layer is a
382    /// rollup layer, io errors may occur.
383    pub async fn triple_removal_exists(
384        &self,
385        subject: u64,
386        predicate: u64,
387        object: u64,
388    ) -> io::Result<bool> {
389        self.store
390            .layer_store
391            .triple_removal_exists(self.layer.name(), subject, predicate, object)
392            .await
393    }
394
395    /// Returns a future that yields an iterator over all layer additions.
396    ///
397    /// Since this operation will involve io when this layer is a
398    /// rollup layer, io errors may occur.
399    pub async fn triple_additions(&self) -> io::Result<Box<dyn Iterator<Item = IdTriple> + Send>> {
400        let result = self
401            .store
402            .layer_store
403            .triple_additions(self.layer.name())
404            .await?;
405
406        Ok(Box::new(result) as Box<dyn Iterator<Item = _> + Send>)
407    }
408
409    /// Returns a future that yields an iterator over all layer removals.
410    ///
411    /// Since this operation will involve io when this layer is a
412    /// rollup layer, io errors may occur.
413    pub async fn triple_removals(&self) -> io::Result<Box<dyn Iterator<Item = IdTriple> + Send>> {
414        let result = self
415            .store
416            .layer_store
417            .triple_removals(self.layer.name())
418            .await?;
419
420        Ok(Box::new(result) as Box<dyn Iterator<Item = _> + Send>)
421    }
422
423    /// Returns a future that yields an iterator over all layer additions that share a particular subject.
424    ///
425    /// Since this operation will involve io when this layer is a
426    /// rollup layer, io errors may occur.
427    pub async fn triple_additions_s(
428        &self,
429        subject: u64,
430    ) -> io::Result<Box<dyn Iterator<Item = IdTriple> + Send>> {
431        self.store
432            .layer_store
433            .triple_additions_s(self.layer.name(), subject)
434            .await
435    }
436
437    /// Returns a future that yields an iterator over all layer removals that share a particular subject.
438    ///
439    /// Since this operation will involve io when this layer is a
440    /// rollup layer, io errors may occur.
441    pub async fn triple_removals_s(
442        &self,
443        subject: u64,
444    ) -> io::Result<Box<dyn Iterator<Item = IdTriple> + Send>> {
445        self.store
446            .layer_store
447            .triple_removals_s(self.layer.name(), subject)
448            .await
449    }
450
451    /// Returns a future that yields an iterator over all layer additions that share a particular subject and predicate.
452    ///
453    /// Since this operation will involve io when this layer is a
454    /// rollup layer, io errors may occur.
455    pub async fn triple_additions_sp(
456        &self,
457        subject: u64,
458        predicate: u64,
459    ) -> io::Result<Box<dyn Iterator<Item = IdTriple> + Send>> {
460        self.store
461            .layer_store
462            .triple_additions_sp(self.layer.name(), subject, predicate)
463            .await
464    }
465
466    /// Returns a future that yields an iterator over all layer removals that share a particular subject and predicate.
467    ///
468    /// Since this operation will involve io when this layer is a
469    /// rollup layer, io errors may occur.
470    pub async fn triple_removals_sp(
471        &self,
472        subject: u64,
473        predicate: u64,
474    ) -> io::Result<Box<dyn Iterator<Item = IdTriple> + Send>> {
475        self.store
476            .layer_store
477            .triple_removals_sp(self.layer.name(), subject, predicate)
478            .await
479    }
480
481    /// Returns a future that yields an iterator over all layer additions that share a particular predicate.
482    ///
483    /// Since this operation will involve io when this layer is a
484    /// rollup layer, io errors may occur.
485    pub async fn triple_additions_p(
486        &self,
487        predicate: u64,
488    ) -> io::Result<Box<dyn Iterator<Item = IdTriple> + Send>> {
489        self.store
490            .layer_store
491            .triple_additions_p(self.layer.name(), predicate)
492            .await
493    }
494
495    /// Returns a future that yields an iterator over all layer removals that share a particular predicate.
496    ///
497    /// Since this operation will involve io when this layer is a
498    /// rollup layer, io errors may occur.
499    pub async fn triple_removals_p(
500        &self,
501        predicate: u64,
502    ) -> io::Result<Box<dyn Iterator<Item = IdTriple> + Send>> {
503        self.store
504            .layer_store
505            .triple_removals_p(self.layer.name(), predicate)
506            .await
507    }
508
509    /// Returns a future that yields an iterator over all layer additions that share a particular object.
510    ///
511    /// Since this operation will involve io when this layer is a
512    /// rollup layer, io errors may occur.
513    pub async fn triple_additions_o(
514        &self,
515        object: u64,
516    ) -> io::Result<Box<dyn Iterator<Item = IdTriple> + Send>> {
517        self.store
518            .layer_store
519            .triple_additions_o(self.layer.name(), object)
520            .await
521    }
522
523    /// Returns a future that yields an iterator over all layer removals that share a particular object.
524    ///
525    /// Since this operation will involve io when this layer is a
526    /// rollup layer, io errors may occur.
527    pub async fn triple_removals_o(
528        &self,
529        object: u64,
530    ) -> io::Result<Box<dyn Iterator<Item = IdTriple> + Send>> {
531        self.store
532            .layer_store
533            .triple_removals_o(self.layer.name(), object)
534            .await
535    }
536
537    /// Returns a future that yields the amount of triples that this layer adds.
538    ///
539    /// Since this operation will involve io when this layer is a
540    /// rollup layer, io errors may occur.
541    pub async fn triple_layer_addition_count(&self) -> io::Result<usize> {
542        self.store
543            .layer_store
544            .triple_layer_addition_count(self.layer.name())
545            .await
546    }
547
548    /// Returns a future that yields the amount of triples that this layer removes.
549    ///
550    /// Since this operation will involve io when this layer is a
551    /// rollup layer, io errors may occur.
552    pub async fn triple_layer_removal_count(&self) -> io::Result<usize> {
553        self.store
554            .layer_store
555            .triple_layer_removal_count(self.layer.name())
556            .await
557    }
558
559    /// Returns a future that yields a vector of layer stack names describing the history of this layer, starting from the base layer up to and including the name of this layer itself.
560    pub async fn retrieve_layer_stack_names(&self) -> io::Result<Vec<[u32; 5]>> {
561        self.store
562            .layer_store
563            .retrieve_layer_stack_names(self.name())
564            .await
565    }
566}
567
568impl PartialEq for StoreLayer {
569    #[allow(clippy::vtable_address_comparisons)]
570    fn eq(&self, other: &StoreLayer) -> bool {
571        Arc::ptr_eq(&self.layer, &other.layer)
572    }
573}
574
575impl Eq for StoreLayer {}
576
577#[async_trait]
578impl Layer for StoreLayer {
579    fn name(&self) -> [u32; 5] {
580        self.layer.name()
581    }
582
583    fn parent_name(&self) -> Option<[u32; 5]> {
584        self.layer.parent_name()
585    }
586
587    fn node_and_value_count(&self) -> usize {
588        self.layer.node_and_value_count()
589    }
590
591    fn predicate_count(&self) -> usize {
592        self.layer.predicate_count()
593    }
594
595    fn subject_id(&self, subject: &str) -> Option<u64> {
596        self.layer.subject_id(subject)
597    }
598
599    fn predicate_id(&self, predicate: &str) -> Option<u64> {
600        self.layer.predicate_id(predicate)
601    }
602
603    fn object_node_id(&self, object: &str) -> Option<u64> {
604        self.layer.object_node_id(object)
605    }
606
607    fn object_value_id(&self, object: &TypedDictEntry) -> Option<u64> {
608        self.layer.object_value_id(object)
609    }
610
611    fn id_subject(&self, id: u64) -> Option<String> {
612        self.layer.id_subject(id)
613    }
614
615    fn id_predicate(&self, id: u64) -> Option<String> {
616        self.layer.id_predicate(id)
617    }
618
619    fn id_object(&self, id: u64) -> Option<ObjectType> {
620        self.layer.id_object(id)
621    }
622
623    fn id_object_is_node(&self, id: u64) -> Option<bool> {
624        self.layer.id_object_is_node(id)
625    }
626
627    fn triple_exists(&self, subject: u64, predicate: u64, object: u64) -> bool {
628        self.layer.triple_exists(subject, predicate, object)
629    }
630
631    fn triples(&self) -> Box<dyn Iterator<Item = IdTriple> + Send> {
632        self.layer.triples()
633    }
634
635    fn triples_s(&self, subject: u64) -> Box<dyn Iterator<Item = IdTriple> + Send> {
636        self.layer.triples_s(subject)
637    }
638
639    fn triples_sp(
640        &self,
641        subject: u64,
642        predicate: u64,
643    ) -> Box<dyn Iterator<Item = IdTriple> + Send> {
644        self.layer.triples_sp(subject, predicate)
645    }
646
647    fn triples_p(&self, predicate: u64) -> Box<dyn Iterator<Item = IdTriple> + Send> {
648        self.layer.triples_p(predicate)
649    }
650
651    fn triples_o(&self, object: u64) -> Box<dyn Iterator<Item = IdTriple> + Send> {
652        self.layer.triples_o(object)
653    }
654
655    fn clone_boxed(&self) -> Box<dyn Layer> {
656        Box::new(self.clone())
657    }
658
659    fn triple_addition_count(&self) -> usize {
660        self.layer.triple_addition_count()
661    }
662
663    fn triple_removal_count(&self) -> usize {
664        self.layer.triple_removal_count()
665    }
666
667    fn all_counts(&self) -> LayerCounts {
668        self.layer.all_counts()
669    }
670
671    fn single_triple_sp(&self, subject: u64, predicate: u64) -> Option<IdTriple> {
672        self.layer.single_triple_sp(subject, predicate)
673    }
674}
675
676/// A named graph in terminus-store.
677///
678/// Named graphs in terminus-store are basically just a label pointing
679/// to a layer. Opening a read transaction to a named graph is just
680/// getting hold of the layer it points at, as layers are
681/// read-only. Writing to a named graph is just making it point to a
682/// new layer.
683#[derive(Clone)]
684pub struct NamedGraph {
685    label: String,
686    store: Store,
687}
688
689impl NamedGraph {
690    fn new(label: String, store: Store) -> Self {
691        NamedGraph { label, store }
692    }
693
694    /// Returns the label name itself.
695    pub fn name(&self) -> &str {
696        &self.label
697    }
698
699    /// Returns the layer this database points at, as well as the label version.
700    pub async fn head_version(&self) -> io::Result<(Option<StoreLayer>, u64)> {
701        let new_label = self.store.label_store.get_label(&self.label).await?;
702
703        match new_label {
704            None => Err(io::Error::new(
705                io::ErrorKind::NotFound,
706                "database not found",
707            )),
708            Some(new_label) => {
709                let layer = match new_label.layer {
710                    None => None,
711                    Some(layer) => {
712                        let layer = self.store.layer_store.get_layer(layer).await?;
713                        match layer {
714                            None => {
715                                return Err(io::Error::new(
716                                    io::ErrorKind::NotFound,
717                                    "layer not found even though it is pointed at by a label",
718                                ))
719                            }
720                            Some(layer) => Some(StoreLayer::wrap(layer, self.store.clone())),
721                        }
722                    }
723                };
724                Ok((layer, new_label.version))
725            }
726        }
727    }
728
729    /// Returns the layer this database points at.
730    pub async fn head(&self) -> io::Result<Option<StoreLayer>> {
731        Ok(self.head_version().await?.0)
732    }
733
734    /// Set the database label to the given layer if it is a valid ancestor, returning false otherwise.
735    pub async fn set_head(&self, layer: &StoreLayer) -> io::Result<bool> {
736        let layer_name = layer.name();
737        let label = self.store.label_store.get_label(&self.label).await?;
738        if label.is_none() {
739            return Err(io::Error::new(io::ErrorKind::NotFound, "label not found"));
740        }
741        let label = label.unwrap();
742
743        let set_is_ok = match label.layer {
744            None => true,
745            Some(retrieved_layer_name) => {
746                self.store
747                    .layer_store
748                    .layer_is_ancestor_of(layer_name, retrieved_layer_name)
749                    .await?
750            }
751        };
752
753        if set_is_ok {
754            Ok(self
755                .store
756                .label_store
757                .set_label(&label, layer_name)
758                .await?
759                .is_some())
760        } else {
761            Ok(false)
762        }
763    }
764
765    /// Set the database label to the given layer, even if it is not a valid ancestor.
766    pub async fn force_set_head(&self, layer: &StoreLayer) -> io::Result<()> {
767        let layer_name = layer.name();
768
769        // We are stomping on the label but `set_label` expects us to
770        // know about the current label, which may have been updated
771        // concurrently.
772        // So keep looping until an update was succesful or an error
773        // was encountered.
774        loop {
775            let label = self.store.label_store.get_label(&self.label).await?;
776            match label {
777                None => return Err(io::Error::new(io::ErrorKind::NotFound, "label not found")),
778                Some(label) => {
779                    if self
780                        .store
781                        .label_store
782                        .set_label(&label, layer_name)
783                        .await?
784                        .is_some()
785                    {
786                        return Ok(());
787                    }
788                }
789            }
790        }
791    }
792
793    /// Set the database label to the given layer, even if it is not a valid ancestor. Also checks given version, and if it doesn't match, the update won't happen and false will be returned.
794    pub async fn force_set_head_version(
795        &self,
796        layer: &StoreLayer,
797        version: u64,
798    ) -> io::Result<bool> {
799        let layer_name = layer.name();
800        let label = self.store.label_store.get_label(&self.label).await?;
801        match label {
802            None => Err(io::Error::new(io::ErrorKind::NotFound, "label not found")),
803            Some(label) => {
804                if label.version != version {
805                    Ok(false)
806                } else {
807                    Ok(self
808                        .store
809                        .label_store
810                        .set_label(&label, layer_name)
811                        .await?
812                        .is_some())
813                }
814            }
815        }
816    }
817
818    pub async fn delete(&self) -> io::Result<()> {
819        self.store.delete(&self.label).await.map(|_| ())
820    }
821}
822
823impl Store {
824    /// Create a new store from the given label and layer store.
825    pub fn new<Labels: 'static + LabelStore, Layers: 'static + LayerStore>(
826        label_store: Labels,
827        layer_store: Layers,
828    ) -> Store {
829        Store {
830            label_store: Arc::new(label_store),
831            layer_store: Arc::new(layer_store),
832        }
833    }
834
835    /// Create a new database with the given name.
836    ///
837    /// If the database already exists, this will return an error.
838    pub async fn create(&self, label: &str) -> io::Result<NamedGraph> {
839        let label = self.label_store.create_label(label).await?;
840        Ok(NamedGraph::new(label.name, self.clone()))
841    }
842
843    /// Open an existing database with the given name, or None if it does not exist.
844    pub async fn open(&self, label: &str) -> io::Result<Option<NamedGraph>> {
845        let label = self.label_store.get_label(label).await?;
846        Ok(label.map(|label| NamedGraph::new(label.name, self.clone())))
847    }
848
849    /// Delete an existing database with the given name. Returns true if this database was deleted
850    /// and false otherwise.
851    pub async fn delete(&self, label: &str) -> io::Result<bool> {
852        self.label_store.delete_label(label).await
853    }
854
855    /// Return list of names of all existing databases.
856    pub async fn labels(&self) -> io::Result<Vec<String>> {
857        let labels = self.label_store.labels().await?;
858        Ok(labels.iter().map(|label| label.name.to_string()).collect())
859    }
860
861    /// Retrieve a layer with the given name from the layer store this Store was initialized with.
862    pub async fn get_layer_from_id(&self, layer: [u32; 5]) -> io::Result<Option<StoreLayer>> {
863        let layer = self.layer_store.get_layer(layer).await?;
864        Ok(layer.map(|layer| StoreLayer::wrap(layer, self.clone())))
865    }
866
867    /// Create a base layer builder, unattached to any database label.
868    ///
869    /// After having committed it, use `set_head` on a `NamedGraph` to attach it.
870    pub async fn create_base_layer(&self) -> io::Result<StoreLayerBuilder> {
871        StoreLayerBuilder::new(self.clone()).await
872    }
873
874    pub async fn merge_base_layers(
875        &self,
876        layers: &[[u32; 5]],
877        temp_dir: &Path,
878    ) -> io::Result<[u32; 5]> {
879        self.layer_store.merge_base_layer(layers, temp_dir).await
880    }
881
882    /// Export the given layers by creating a pack, a Vec<u8> that can later be used with `import_layers` on a different store.
883    pub async fn export_layers(
884        &self,
885        layer_ids: Box<dyn Iterator<Item = [u32; 5]> + Send>,
886    ) -> io::Result<Vec<u8>> {
887        self.layer_store.export_layers(layer_ids).await
888    }
889
890    /// Import the specified layers from the given pack, a byte slice that was previously generated with `export_layers`, on another store, and possibly even another machine).
891    ///
892    /// After this operation, the specified layers will be retrievable
893    /// from this store, provided they existed in the pack. specified
894    /// layers that are not in the pack are silently ignored.
895    pub async fn import_layers<'a>(
896        &'a self,
897        pack: &'a [u8],
898        layer_ids: Box<dyn Iterator<Item = [u32; 5]> + Send>,
899    ) -> io::Result<()> {
900        self.layer_store.import_layers(pack, layer_ids).await
901    }
902}
903
904/// Open a store that is entirely in memory.
905///
906/// This is useful for testing purposes, or if the database is only going to be used for caching purposes.
907pub fn open_memory_store() -> Store {
908    Store::new(
909        MemoryLabelStore::new(),
910        CachedLayerStore::new(MemoryLayerStore::new(), LockingHashMapLayerCache::new()),
911    )
912}
913
914/// Open a store that stores its data in the given directory as archive files.
915///
916/// cache_size specifies in megabytes how large the LRU cache should
917/// be. Loaded layers will stick around in the LRU cache to speed up
918/// subsequent loads.
919pub fn open_archive_store<P: Into<PathBuf>>(path: P, cache_size: usize) -> Store {
920    let p = path.into();
921    let directory_archive_backend = DirectoryArchiveBackend::new(p.clone());
922    let archive_backend = LruArchiveBackend::new(
923        directory_archive_backend.clone(),
924        directory_archive_backend,
925        cache_size,
926    );
927    Store::new(
928        DirectoryLabelStore::new(p),
929        CachedLayerStore::new(
930            ArchiveLayerStore::new(archive_backend.clone(), archive_backend),
931            LockingHashMapLayerCache::new(),
932        ),
933    )
934}
935
936/// Open a store that stores its data in the given directory as archive files.
937///
938/// This version doesn't use lru caching.
939pub fn open_raw_archive_store<P: Into<PathBuf>>(path: P) -> Store {
940    let p = path.into();
941    let archive_backend = DirectoryArchiveBackend::new(p.clone());
942    Store::new(
943        DirectoryLabelStore::new(p),
944        CachedLayerStore::new(
945            ArchiveLayerStore::new(archive_backend.clone(), archive_backend),
946            LockingHashMapLayerCache::new(),
947        ),
948    )
949}
950
951/// Open a store that stores its data in the given directory.
952pub fn open_directory_store<P: Into<PathBuf>>(path: P) -> Store {
953    let p = path.into();
954    Store::new(
955        DirectoryLabelStore::new(p.clone()),
956        CachedLayerStore::new(DirectoryLayerStore::new(p), LockingHashMapLayerCache::new()),
957    )
958}
959
960#[cfg(test)]
961mod tests {
962    use super::*;
963    use tempfile::tempdir;
964
965    async fn create_and_manipulate_database(store: Store) {
966        let database = store.create("foodb").await.unwrap();
967
968        let head = database.head().await.unwrap();
969        assert!(head.is_none());
970
971        let mut builder = store.create_base_layer().await.unwrap();
972        builder
973            .add_value_triple(ValueTriple::new_string_value("cow", "says", "moo"))
974            .unwrap();
975
976        let layer = builder.commit().await.unwrap();
977        assert!(database.set_head(&layer).await.unwrap());
978
979        builder = layer.open_write().await.unwrap();
980        builder
981            .add_value_triple(ValueTriple::new_string_value("pig", "says", "oink"))
982            .unwrap();
983
984        let layer2 = builder.commit().await.unwrap();
985        assert!(database.set_head(&layer2).await.unwrap());
986        let layer2_name = layer2.name();
987
988        let layer = database.head().await.unwrap().unwrap();
989
990        assert_eq!(layer2_name, layer.name());
991        assert!(layer.value_triple_exists(&ValueTriple::new_string_value("cow", "says", "moo")));
992        assert!(layer.value_triple_exists(&ValueTriple::new_string_value("pig", "says", "oink")));
993    }
994
995    #[tokio::test]
996    async fn create_and_manipulate_memory_database() {
997        let store = open_memory_store();
998
999        create_and_manipulate_database(store).await;
1000    }
1001
1002    #[tokio::test]
1003    async fn create_and_manipulate_directory_database() {
1004        let dir = tempdir().unwrap();
1005        let store = open_directory_store(dir.path());
1006
1007        create_and_manipulate_database(store).await;
1008    }
1009
1010    #[tokio::test]
1011    async fn create_layer_and_retrieve_it_by_id() {
1012        let store = open_memory_store();
1013        let builder = store.create_base_layer().await.unwrap();
1014        builder
1015            .add_value_triple(ValueTriple::new_string_value("cow", "says", "moo"))
1016            .unwrap();
1017
1018        let layer = builder.commit().await.unwrap();
1019
1020        let id = layer.name();
1021
1022        let layer2 = store.get_layer_from_id(id).await.unwrap().unwrap();
1023
1024        assert!(layer2.value_triple_exists(&ValueTriple::new_string_value("cow", "says", "moo")));
1025    }
1026
1027    #[tokio::test]
1028    async fn commit_builder_makes_builder_committed() {
1029        let store = open_memory_store();
1030        let builder = store.create_base_layer().await.unwrap();
1031
1032        builder
1033            .add_value_triple(ValueTriple::new_string_value("cow", "says", "moo"))
1034            .unwrap();
1035
1036        assert!(!builder.committed());
1037
1038        builder.commit_no_load().await.unwrap();
1039
1040        assert!(builder.committed());
1041    }
1042
1043    #[tokio::test]
1044    async fn hard_reset() {
1045        let store = open_memory_store();
1046        let database = store.create("foodb").await.unwrap();
1047
1048        let builder1 = store.create_base_layer().await.unwrap();
1049        builder1
1050            .add_value_triple(ValueTriple::new_string_value("cow", "says", "moo"))
1051            .unwrap();
1052
1053        let layer1 = builder1.commit().await.unwrap();
1054
1055        assert!(database.set_head(&layer1).await.unwrap());
1056
1057        let builder2 = store.create_base_layer().await.unwrap();
1058        builder2
1059            .add_value_triple(ValueTriple::new_string_value("duck", "says", "quack"))
1060            .unwrap();
1061
1062        let layer2 = builder2.commit().await.unwrap();
1063
1064        database.force_set_head(&layer2).await.unwrap();
1065
1066        let new_layer = database.head().await.unwrap().unwrap();
1067
1068        assert!(
1069            new_layer.value_triple_exists(&ValueTriple::new_string_value("duck", "says", "quack"))
1070        );
1071        assert!(
1072            !new_layer.value_triple_exists(&ValueTriple::new_string_value("cow", "says", "moo"))
1073        );
1074    }
1075
1076    #[tokio::test]
1077    async fn create_two_layers_and_squash() {
1078        let store = open_memory_store();
1079        let builder = store.create_base_layer().await.unwrap();
1080        builder
1081            .add_value_triple(ValueTriple::new_string_value("cow", "says", "moo"))
1082            .unwrap();
1083        builder
1084            .add_value_triple(ValueTriple::new_node("cow", "likes", "duck"))
1085            .unwrap();
1086        builder
1087            .add_value_triple(ValueTriple::new_node("cow", "likes", "horse"))
1088            .unwrap();
1089
1090        let layer = builder.commit().await.unwrap();
1091
1092        let builder2 = layer.open_write().await.unwrap();
1093
1094        builder2
1095            .add_value_triple(ValueTriple::new_string_value("dog", "says", "woof"))
1096            .unwrap();
1097
1098        builder2
1099            .add_value_triple(ValueTriple::new_string_value("bunny", "says", "sniff"))
1100            .unwrap();
1101
1102        builder2
1103            .remove_value_triple(ValueTriple::new_string_value("cow", "says", "moo"))
1104            .unwrap();
1105
1106        builder2
1107            .remove_value_triple(ValueTriple::new_node("cow", "likes", "horse"))
1108            .unwrap();
1109
1110        builder2
1111            .add_value_triple(ValueTriple::new_node("bunny", "likes", "cow"))
1112            .unwrap();
1113
1114        builder2
1115            .add_value_triple(ValueTriple::new_node("cow", "likes", "duck"))
1116            .unwrap();
1117
1118        let layer2 = builder2.commit().await.unwrap();
1119
1120        let new = layer2.squash().await.unwrap();
1121        let triples: Vec<_> = new
1122            .triples()
1123            .map(|t| new.id_triple_to_string(&t).unwrap())
1124            .collect();
1125        assert_eq!(
1126            vec![
1127                ValueTriple::new_node("bunny", "likes", "cow"),
1128                ValueTriple::new_string_value("bunny", "says", "sniff"),
1129                ValueTriple::new_node("cow", "likes", "duck"),
1130                ValueTriple::new_string_value("dog", "says", "woof"),
1131            ],
1132            triples
1133        );
1134
1135        assert!(new.parent().await.unwrap().is_none());
1136    }
1137
1138    #[tokio::test]
1139    async fn create_three_layers_and_squash_last_two() {
1140        let store = open_memory_store();
1141        let builder = store.create_base_layer().await.unwrap();
1142        builder
1143            .add_value_triple(ValueTriple::new_string_value("cow", "says", "quack"))
1144            .unwrap();
1145        builder
1146            .add_value_triple(ValueTriple::new_node("cow", "hates", "duck"))
1147            .unwrap();
1148        builder
1149            .add_value_triple(ValueTriple::new_node("cow", "likes", "horse"))
1150            .unwrap();
1151
1152        let base_layer = builder.commit().await.unwrap();
1153
1154        let builder = base_layer.open_write().await.unwrap();
1155        builder
1156            .add_value_triple(ValueTriple::new_node("bunny", "likes", "cow"))
1157            .unwrap();
1158        builder
1159            .add_value_triple(ValueTriple::new_string_value("bunny", "says", "neigh"))
1160            .unwrap();
1161        builder
1162            .add_value_triple(ValueTriple::new_node("duck", "likes", "cow"))
1163            .unwrap();
1164        builder
1165            .add_value_triple(ValueTriple::new_string_value("duck", "says", "quack"))
1166            .unwrap();
1167        builder
1168            .remove_value_triple(ValueTriple::new_string_value("cow", "says", "quack"))
1169            .unwrap();
1170
1171        let intermediate_layer = builder.commit().await.unwrap();
1172        let builder = intermediate_layer.open_write().await.unwrap();
1173        builder
1174            .remove_value_triple(ValueTriple::new_node("cow", "hates", "duck"))
1175            .unwrap();
1176        builder
1177            .remove_value_triple(ValueTriple::new_string_value("bunny", "says", "neigh"))
1178            .unwrap();
1179        builder
1180            .add_value_triple(ValueTriple::new_node("cow", "likes", "duck"))
1181            .unwrap();
1182        builder
1183            .add_value_triple(ValueTriple::new_string_value("cow", "says", "moo"))
1184            .unwrap();
1185        builder
1186            .add_value_triple(ValueTriple::new_string_value("bunny", "says", "sniff"))
1187            .unwrap();
1188        let final_layer = builder.commit().await.unwrap();
1189
1190        let squashed_layer = final_layer.squash_upto(&base_layer).await.unwrap();
1191        assert_eq!(squashed_layer.parent_name().unwrap(), base_layer.name());
1192        let additions: Vec<_> = squashed_layer
1193            .triple_additions()
1194            .await
1195            .unwrap()
1196            .map(|t| squashed_layer.id_triple_to_string(&t).unwrap())
1197            .collect();
1198        assert_eq!(
1199            vec![
1200                ValueTriple::new_node("cow", "likes", "duck"),
1201                ValueTriple::new_string_value("cow", "says", "moo"),
1202                ValueTriple::new_node("duck", "likes", "cow"),
1203                ValueTriple::new_string_value("duck", "says", "quack"),
1204                ValueTriple::new_node("bunny", "likes", "cow"),
1205                ValueTriple::new_string_value("bunny", "says", "sniff"),
1206            ],
1207            additions
1208        );
1209        let removals: Vec<_> = squashed_layer
1210            .triple_removals()
1211            .await
1212            .unwrap()
1213            .map(|t| squashed_layer.id_triple_to_string(&t).unwrap())
1214            .collect();
1215        assert_eq!(
1216            vec![
1217                ValueTriple::new_node("cow", "hates", "duck"),
1218                ValueTriple::new_string_value("cow", "says", "quack"),
1219            ],
1220            removals
1221        );
1222
1223        let all_triples: Vec<_> = squashed_layer
1224            .triples()
1225            .map(|t| squashed_layer.id_triple_to_string(&t).unwrap())
1226            .collect();
1227        assert_eq!(
1228            vec![
1229                ValueTriple::new_node("cow", "likes", "duck"),
1230                ValueTriple::new_node("cow", "likes", "horse"),
1231                ValueTriple::new_string_value("cow", "says", "moo"),
1232                ValueTriple::new_node("duck", "likes", "cow"),
1233                ValueTriple::new_string_value("duck", "says", "quack"),
1234                ValueTriple::new_node("bunny", "likes", "cow"),
1235                ValueTriple::new_string_value("bunny", "says", "sniff"),
1236            ],
1237            all_triples
1238        );
1239    }
1240
1241    #[tokio::test]
1242    async fn create_three_layers_and_squash_all_after_rollup() {
1243        let store = open_memory_store();
1244        let builder = store.create_base_layer().await.unwrap();
1245        builder
1246            .add_value_triple(ValueTriple::new_string_value("cow", "says", "quack"))
1247            .unwrap();
1248        builder
1249            .add_value_triple(ValueTriple::new_node("cow", "hates", "duck"))
1250            .unwrap();
1251        builder
1252            .add_value_triple(ValueTriple::new_node("cow", "likes", "horse"))
1253            .unwrap();
1254
1255        let base_layer = builder.commit().await.unwrap();
1256
1257        let builder = base_layer.open_write().await.unwrap();
1258        builder
1259            .add_value_triple(ValueTriple::new_node("bunny", "likes", "cow"))
1260            .unwrap();
1261        builder
1262            .add_value_triple(ValueTriple::new_string_value("bunny", "says", "neigh"))
1263            .unwrap();
1264        builder
1265            .add_value_triple(ValueTriple::new_node("duck", "likes", "cow"))
1266            .unwrap();
1267        builder
1268            .add_value_triple(ValueTriple::new_string_value("duck", "says", "quack"))
1269            .unwrap();
1270        builder
1271            .remove_value_triple(ValueTriple::new_string_value("cow", "says", "quack"))
1272            .unwrap();
1273
1274        let intermediate_layer = builder.commit().await.unwrap();
1275        let builder = intermediate_layer.open_write().await.unwrap();
1276        builder
1277            .remove_value_triple(ValueTriple::new_node("cow", "hates", "duck"))
1278            .unwrap();
1279        builder
1280            .remove_value_triple(ValueTriple::new_string_value("bunny", "says", "neigh"))
1281            .unwrap();
1282        builder
1283            .add_value_triple(ValueTriple::new_node("cow", "likes", "duck"))
1284            .unwrap();
1285        builder
1286            .add_value_triple(ValueTriple::new_string_value("cow", "says", "moo"))
1287            .unwrap();
1288        builder
1289            .add_value_triple(ValueTriple::new_string_value("bunny", "says", "sniff"))
1290            .unwrap();
1291        let final_layer = builder.commit().await.unwrap();
1292        final_layer.rollup_upto(&base_layer).await.unwrap();
1293        let final_rolled_layer = store
1294            .get_layer_from_id(final_layer.name())
1295            .await
1296            .unwrap()
1297            .unwrap();
1298
1299        let squashed_layer = final_rolled_layer.squash().await.unwrap();
1300        assert!(squashed_layer.parent_name().is_none());
1301
1302        let all_triples: Vec<_> = squashed_layer
1303            .triples()
1304            .map(|t| squashed_layer.id_triple_to_string(&t).unwrap())
1305            .collect();
1306        assert_eq!(
1307            vec![
1308                ValueTriple::new_node("bunny", "likes", "cow"),
1309                ValueTriple::new_string_value("bunny", "says", "sniff"),
1310                ValueTriple::new_node("cow", "likes", "duck"),
1311                ValueTriple::new_node("cow", "likes", "horse"),
1312                ValueTriple::new_string_value("cow", "says", "moo"),
1313                ValueTriple::new_node("duck", "likes", "cow"),
1314                ValueTriple::new_string_value("duck", "says", "quack"),
1315            ],
1316            all_triples
1317        );
1318    }
1319
1320    #[tokio::test]
1321    async fn create_three_layers_and_squash_last_two_after_rollup() {
1322        let store = open_memory_store();
1323        let builder = store.create_base_layer().await.unwrap();
1324        builder
1325            .add_value_triple(ValueTriple::new_string_value("cow", "says", "quack"))
1326            .unwrap();
1327        builder
1328            .add_value_triple(ValueTriple::new_node("cow", "hates", "duck"))
1329            .unwrap();
1330        builder
1331            .add_value_triple(ValueTriple::new_node("cow", "likes", "horse"))
1332            .unwrap();
1333
1334        let base_layer = builder.commit().await.unwrap();
1335
1336        let builder = base_layer.open_write().await.unwrap();
1337        builder
1338            .add_value_triple(ValueTriple::new_node("bunny", "likes", "cow"))
1339            .unwrap();
1340        builder
1341            .add_value_triple(ValueTriple::new_string_value("bunny", "says", "neigh"))
1342            .unwrap();
1343        builder
1344            .add_value_triple(ValueTriple::new_node("duck", "likes", "cow"))
1345            .unwrap();
1346        builder
1347            .add_value_triple(ValueTriple::new_string_value("duck", "says", "quack"))
1348            .unwrap();
1349        builder
1350            .remove_value_triple(ValueTriple::new_string_value("cow", "says", "quack"))
1351            .unwrap();
1352
1353        let intermediate_layer = builder.commit().await.unwrap();
1354        let builder = intermediate_layer.open_write().await.unwrap();
1355        builder
1356            .remove_value_triple(ValueTriple::new_node("cow", "hates", "duck"))
1357            .unwrap();
1358        builder
1359            .remove_value_triple(ValueTriple::new_string_value("bunny", "says", "neigh"))
1360            .unwrap();
1361        builder
1362            .add_value_triple(ValueTriple::new_node("cow", "likes", "duck"))
1363            .unwrap();
1364        builder
1365            .add_value_triple(ValueTriple::new_string_value("cow", "says", "moo"))
1366            .unwrap();
1367        builder
1368            .add_value_triple(ValueTriple::new_string_value("bunny", "says", "sniff"))
1369            .unwrap();
1370        let final_layer = builder.commit().await.unwrap();
1371        final_layer.rollup_upto(&base_layer).await.unwrap();
1372        let final_rolled_layer = store
1373            .get_layer_from_id(final_layer.name())
1374            .await
1375            .unwrap()
1376            .unwrap();
1377
1378        let squashed_layer = final_rolled_layer.squash_upto(&base_layer).await.unwrap();
1379        assert_eq!(squashed_layer.parent_name().unwrap(), base_layer.name());
1380        let additions: Vec<_> = squashed_layer
1381            .triple_additions()
1382            .await
1383            .unwrap()
1384            .map(|t| squashed_layer.id_triple_to_string(&t).unwrap())
1385            .collect();
1386        assert_eq!(
1387            vec![
1388                ValueTriple::new_node("cow", "likes", "duck"),
1389                ValueTriple::new_string_value("cow", "says", "moo"),
1390                ValueTriple::new_node("duck", "likes", "cow"),
1391                ValueTriple::new_string_value("duck", "says", "quack"),
1392                ValueTriple::new_node("bunny", "likes", "cow"),
1393                ValueTriple::new_string_value("bunny", "says", "sniff"),
1394            ],
1395            additions
1396        );
1397        let removals: Vec<_> = squashed_layer
1398            .triple_removals()
1399            .await
1400            .unwrap()
1401            .map(|t| squashed_layer.id_triple_to_string(&t).unwrap())
1402            .collect();
1403        assert_eq!(
1404            vec![
1405                ValueTriple::new_node("cow", "hates", "duck"),
1406                ValueTriple::new_string_value("cow", "says", "quack"),
1407            ],
1408            removals
1409        );
1410
1411        let all_triples: Vec<_> = squashed_layer
1412            .triples()
1413            .map(|t| squashed_layer.id_triple_to_string(&t).unwrap())
1414            .collect();
1415        assert_eq!(
1416            vec![
1417                ValueTriple::new_node("cow", "likes", "duck"),
1418                ValueTriple::new_node("cow", "likes", "horse"),
1419                ValueTriple::new_string_value("cow", "says", "moo"),
1420                ValueTriple::new_node("duck", "likes", "cow"),
1421                ValueTriple::new_string_value("duck", "says", "quack"),
1422                ValueTriple::new_node("bunny", "likes", "cow"),
1423                ValueTriple::new_string_value("bunny", "says", "sniff"),
1424            ],
1425            all_triples
1426        );
1427    }
1428
1429    #[tokio::test]
1430    async fn squash_and_forget_dict_entries() {
1431        let store = open_memory_store();
1432        let builder = store.create_base_layer().await.unwrap();
1433        builder
1434            .add_value_triple(ValueTriple::new_node("a", "b", "anode"))
1435            .unwrap();
1436        builder
1437            .add_value_triple(ValueTriple::new_string_value("a", "b", "astring"))
1438            .unwrap();
1439        builder
1440            .add_value_triple(ValueTriple::new_node("a", "c", "anothernode"))
1441            .unwrap();
1442        builder
1443            .add_value_triple(ValueTriple::new_string_value("a", "c", "anotherstring"))
1444            .unwrap();
1445
1446        let base_layer = builder.commit().await.unwrap();
1447
1448        let builder = base_layer.open_write().await.unwrap();
1449        builder
1450            .remove_value_triple(ValueTriple::new_node("a", "c", "anothernode"))
1451            .unwrap();
1452        builder
1453            .remove_value_triple(ValueTriple::new_string_value("a", "c", "anotherstring"))
1454            .unwrap();
1455        let child_layer = builder.commit().await.unwrap();
1456
1457        let squashed = child_layer.squash().await.unwrap();
1458        // annoyingly we need to get the internal layer version, so lets re-retrieve
1459        let squashed = store
1460            .layer_store
1461            .get_layer(squashed.name())
1462            .await
1463            .unwrap()
1464            .unwrap();
1465        let nodes: Vec<_> = squashed
1466            .node_dictionary()
1467            .iter()
1468            .map(|b| b.to_bytes())
1469            .collect();
1470        assert_eq!(vec![b"a" as &[u8], b"anode"], nodes);
1471        let preds: Vec<_> = squashed
1472            .predicate_dictionary()
1473            .iter()
1474            .map(|b| b.to_bytes())
1475            .collect();
1476        assert_eq!(vec![b"b" as &[u8]], preds);
1477        let vals: Vec<_> = squashed
1478            .value_dictionary()
1479            .iter()
1480            .map(|b| b.to_bytes())
1481            .collect();
1482        assert_eq!(vec![b"astring" as &[u8]], vals);
1483
1484        let all_triples: Vec<_> = squashed
1485            .triples()
1486            .map(|t| squashed.id_triple_to_string(&t).unwrap())
1487            .collect();
1488        assert_eq!(
1489            vec![
1490                ValueTriple::new_node("a", "b", "anode"),
1491                ValueTriple::new_string_value("a", "b", "astring"),
1492            ],
1493            all_triples
1494        );
1495    }
1496
1497    #[tokio::test]
1498    async fn squash_upto_and_forget_dict_entries() {
1499        let store = open_memory_store();
1500        let builder = store.create_base_layer().await.unwrap();
1501        builder
1502            .add_value_triple(ValueTriple::new_node("foo", "bar", "baz"))
1503            .unwrap();
1504        builder
1505            .add_value_triple(ValueTriple::new_node("baz", "bar", "quux"))
1506            .unwrap();
1507        builder
1508            .add_value_triple(ValueTriple::new_string_value("foo", "baz", "hai"))
1509            .unwrap();
1510        let base_layer = builder.commit().await.unwrap();
1511        let builder = base_layer.open_write().await.unwrap();
1512        builder
1513            .remove_value_triple(ValueTriple::new_string_value("foo", "baz", "hai"))
1514            .unwrap();
1515        builder
1516            .add_value_triple(ValueTriple::new_node("a", "b", "anode"))
1517            .unwrap();
1518        builder
1519            .add_value_triple(ValueTriple::new_string_value("a", "b", "astring"))
1520            .unwrap();
1521        builder
1522            .add_value_triple(ValueTriple::new_node("a", "c", "anothernode"))
1523            .unwrap();
1524        builder
1525            .add_value_triple(ValueTriple::new_string_value("a", "c", "anotherstring"))
1526            .unwrap();
1527
1528        let child_layer1 = builder.commit().await.unwrap();
1529
1530        let builder = child_layer1.open_write().await.unwrap();
1531        builder
1532            .remove_value_triple(ValueTriple::new_node("foo", "bar", "baz"))
1533            .unwrap();
1534        builder
1535            .remove_value_triple(ValueTriple::new_node("a", "c", "anothernode"))
1536            .unwrap();
1537        builder
1538            .remove_value_triple(ValueTriple::new_string_value("a", "c", "anotherstring"))
1539            .unwrap();
1540        let child_layer2 = builder.commit().await.unwrap();
1541
1542        let squashed = child_layer2.squash_upto(&base_layer).await.unwrap();
1543        // annoyingly we need to get the internal layer version, so lets re-retrieve
1544        let squashed = store
1545            .layer_store
1546            .get_layer(squashed.name())
1547            .await
1548            .unwrap()
1549            .unwrap();
1550        let nodes: Vec<_> = squashed
1551            .node_dictionary()
1552            .iter()
1553            .map(|b| b.to_bytes())
1554            .collect();
1555        assert_eq!(vec![b"a" as &[u8], b"anode"], nodes);
1556        let preds: Vec<_> = squashed
1557            .predicate_dictionary()
1558            .iter()
1559            .map(|b| b.to_bytes())
1560            .collect();
1561        assert_eq!(vec![b"b" as &[u8]], preds);
1562        let vals: Vec<_> = squashed
1563            .value_dictionary()
1564            .iter()
1565            .map(|b| b.to_bytes())
1566            .collect();
1567        assert_eq!(vec![b"astring" as &[u8]], vals);
1568
1569        let all_triple_additions: Vec<_> = squashed
1570            .internal_triple_additions()
1571            .map(|t| squashed.id_triple_to_string(&t).unwrap())
1572            .collect();
1573        let all_triple_removals: Vec<_> = squashed
1574            .internal_triple_removals()
1575            .map(|t| squashed.id_triple_to_string(&t).unwrap())
1576            .collect();
1577        assert_eq!(
1578            vec![
1579                ValueTriple::new_node("a", "b", "anode"),
1580                ValueTriple::new_string_value("a", "b", "astring"),
1581            ],
1582            all_triple_additions
1583        );
1584        assert_eq!(
1585            vec![
1586                ValueTriple::new_node("foo", "bar", "baz"),
1587                ValueTriple::new_string_value("foo", "baz", "hai"),
1588            ],
1589            all_triple_removals
1590        );
1591    }
1592
1593    #[tokio::test]
1594    async fn apply_a_base_delta() {
1595        let store = open_memory_store();
1596        let builder = store.create_base_layer().await.unwrap();
1597
1598        builder
1599            .add_value_triple(ValueTriple::new_string_value("cow", "says", "moo"))
1600            .unwrap();
1601
1602        let layer = builder.commit().await.unwrap();
1603
1604        let builder2 = layer.open_write().await.unwrap();
1605
1606        builder2
1607            .add_value_triple(ValueTriple::new_string_value("dog", "says", "woof"))
1608            .unwrap();
1609
1610        let layer2 = builder2.commit().await.unwrap();
1611
1612        let delta_builder_1 = store.create_base_layer().await.unwrap();
1613
1614        delta_builder_1
1615            .add_value_triple(ValueTriple::new_string_value("dog", "says", "woof"))
1616            .unwrap();
1617        delta_builder_1
1618            .add_value_triple(ValueTriple::new_string_value("cat", "says", "meow"))
1619            .unwrap();
1620
1621        let delta_1 = delta_builder_1.commit().await.unwrap();
1622
1623        let delta_builder_2 = delta_1.open_write().await.unwrap();
1624
1625        delta_builder_2
1626            .add_value_triple(ValueTriple::new_string_value("crow", "says", "caw"))
1627            .unwrap();
1628        delta_builder_2
1629            .remove_value_triple(ValueTriple::new_string_value("cat", "says", "meow"))
1630            .unwrap();
1631
1632        let delta = delta_builder_2.commit().await.unwrap();
1633
1634        let rebase_builder = layer2.open_write().await.unwrap();
1635
1636        let _ = rebase_builder.apply_delta(&delta).await.unwrap();
1637
1638        let rebase_layer = rebase_builder.commit().await.unwrap();
1639
1640        assert!(
1641            rebase_layer.value_triple_exists(&ValueTriple::new_string_value("cow", "says", "moo"))
1642        );
1643        assert!(
1644            rebase_layer.value_triple_exists(&ValueTriple::new_string_value("crow", "says", "caw"))
1645        );
1646        assert!(
1647            rebase_layer.value_triple_exists(&ValueTriple::new_string_value("dog", "says", "woof"))
1648        );
1649        assert!(!rebase_layer
1650            .value_triple_exists(&ValueTriple::new_string_value("cat", "says", "meow")));
1651    }
1652
1653    async fn cached_layer_name_does_not_change_after_rollup(store: Store) {
1654        let builder = store.create_base_layer().await.unwrap();
1655        let base_name = builder.name();
1656        let x = builder.commit().await.unwrap();
1657        let builder = x.open_write().await.unwrap();
1658        let child_name = builder.name();
1659        builder.commit().await.unwrap();
1660
1661        let unrolled_layer = store.get_layer_from_id(child_name).await.unwrap().unwrap();
1662        let unrolled_name = unrolled_layer.name();
1663        let unrolled_parent_name = unrolled_layer.parent_name().unwrap();
1664        assert_eq!(child_name, unrolled_name);
1665        assert_eq!(base_name, unrolled_parent_name);
1666
1667        unrolled_layer.rollup().await.unwrap();
1668        let rolled_layer = store.get_layer_from_id(child_name).await.unwrap().unwrap();
1669        let rolled_name = rolled_layer.name();
1670        let rolled_parent_name = rolled_layer.parent_name().unwrap();
1671        assert_eq!(child_name, rolled_name);
1672        assert_eq!(base_name, rolled_parent_name);
1673
1674        rolled_layer.rollup().await.unwrap();
1675        let rolled_layer2 = store.get_layer_from_id(child_name).await.unwrap().unwrap();
1676        let rolled_name2 = rolled_layer2.name();
1677        let rolled_parent_name2 = rolled_layer2.parent_name().unwrap();
1678        assert_eq!(child_name, rolled_name2);
1679        assert_eq!(base_name, rolled_parent_name2);
1680    }
1681
1682    #[tokio::test]
1683    async fn mem_cached_layer_name_does_not_change_after_rollup() {
1684        let store = open_memory_store();
1685
1686        cached_layer_name_does_not_change_after_rollup(store).await
1687    }
1688
1689    #[tokio::test]
1690    async fn dir_cached_layer_name_does_not_change_after_rollup() {
1691        let dir = tempdir().unwrap();
1692        let store = open_directory_store(dir.path());
1693
1694        cached_layer_name_does_not_change_after_rollup(store).await
1695    }
1696
1697    async fn cached_layer_name_does_not_change_after_rollup_upto(store: Store) {
1698        let builder = store.create_base_layer().await.unwrap();
1699        let _base_name = builder.name();
1700        let base_layer = builder.commit().await.unwrap();
1701        let builder = base_layer.open_write().await.unwrap();
1702        let child_name = builder.name();
1703        let x = builder.commit().await.unwrap();
1704        let builder = x.open_write().await.unwrap();
1705        let child_name2 = builder.name();
1706        builder.commit().await.unwrap();
1707
1708        let unrolled_layer = store.get_layer_from_id(child_name2).await.unwrap().unwrap();
1709        let unrolled_name = unrolled_layer.name();
1710        let unrolled_parent_name = unrolled_layer.parent_name().unwrap();
1711        assert_eq!(child_name2, unrolled_name);
1712        assert_eq!(child_name, unrolled_parent_name);
1713
1714        unrolled_layer.rollup_upto(&base_layer).await.unwrap();
1715        let rolled_layer = store.get_layer_from_id(child_name2).await.unwrap().unwrap();
1716        let rolled_name = rolled_layer.name();
1717        let rolled_parent_name = rolled_layer.parent_name().unwrap();
1718        assert_eq!(child_name2, rolled_name);
1719        assert_eq!(child_name, rolled_parent_name);
1720
1721        rolled_layer.rollup_upto(&base_layer).await.unwrap();
1722        let rolled_layer2 = store.get_layer_from_id(child_name2).await.unwrap().unwrap();
1723        let rolled_name2 = rolled_layer2.name();
1724        let rolled_parent_name2 = rolled_layer2.parent_name().unwrap();
1725        assert_eq!(child_name2, rolled_name2);
1726        assert_eq!(child_name, rolled_parent_name2);
1727    }
1728
1729    #[tokio::test]
1730    async fn mem_cached_layer_name_does_not_change_after_rollup_upto() {
1731        let store = open_memory_store();
1732        cached_layer_name_does_not_change_after_rollup_upto(store).await
1733    }
1734
1735    #[tokio::test]
1736    async fn dir_cached_layer_name_does_not_change_after_rollup_upto() {
1737        let dir = tempdir().unwrap();
1738        let store = open_directory_store(dir.path());
1739        cached_layer_name_does_not_change_after_rollup_upto(store).await
1740    }
1741
1742    #[tokio::test]
1743    async fn force_update_with_matching_0_version_succeeds() {
1744        let dir = tempdir().unwrap();
1745        let store = open_directory_store(dir.path());
1746        let graph = store.create("foo").await.unwrap();
1747        let (layer, version) = graph.head_version().await.unwrap();
1748        assert!(layer.is_none());
1749        assert_eq!(0, version);
1750
1751        let builder = store.create_base_layer().await.unwrap();
1752        let layer = builder.commit().await.unwrap();
1753
1754        assert!(graph.force_set_head_version(&layer, 0).await.unwrap());
1755    }
1756
1757    #[tokio::test]
1758    async fn force_update_with_mismatching_0_version_succeeds() {
1759        let dir = tempdir().unwrap();
1760        let store = open_directory_store(dir.path());
1761        let graph = store.create("foo").await.unwrap();
1762        let (layer, version) = graph.head_version().await.unwrap();
1763        assert!(layer.is_none());
1764        assert_eq!(0, version);
1765
1766        let builder = store.create_base_layer().await.unwrap();
1767        let layer = builder.commit().await.unwrap();
1768
1769        assert!(!graph.force_set_head_version(&layer, 3).await.unwrap());
1770    }
1771
1772    #[tokio::test]
1773    async fn force_update_with_matching_version_succeeds() {
1774        let dir = tempdir().unwrap();
1775        let store = open_directory_store(dir.path());
1776        let graph = store.create("foo").await.unwrap();
1777
1778        let builder = store.create_base_layer().await.unwrap();
1779        let layer = builder.commit().await.unwrap();
1780        assert!(graph.set_head(&layer).await.unwrap());
1781
1782        let (_, version) = graph.head_version().await.unwrap();
1783        assert_eq!(1, version);
1784
1785        let builder2 = store.create_base_layer().await.unwrap();
1786        let layer2 = builder2.commit().await.unwrap();
1787
1788        assert!(graph.force_set_head_version(&layer2, 1).await.unwrap());
1789    }
1790
1791    #[tokio::test]
1792    async fn force_update_with_mismatched_version_succeeds() {
1793        let dir = tempdir().unwrap();
1794        let store = open_directory_store(dir.path());
1795        let graph = store.create("foo").await.unwrap();
1796
1797        let builder = store.create_base_layer().await.unwrap();
1798        let layer = builder.commit().await.unwrap();
1799        assert!(graph.set_head(&layer).await.unwrap());
1800
1801        let (_, version) = graph.head_version().await.unwrap();
1802        assert_eq!(1, version);
1803
1804        let builder2 = store.create_base_layer().await.unwrap();
1805        let layer2 = builder2.commit().await.unwrap();
1806
1807        assert!(!graph.force_set_head_version(&layer2, 0).await.unwrap());
1808    }
1809
1810    #[tokio::test]
1811    async fn delete_database() {
1812        let dir = tempdir().unwrap();
1813        let store = open_directory_store(dir.path());
1814        let _ = store.create("foo").await.unwrap();
1815        assert!(store.delete("foo").await.unwrap());
1816        assert!(store.open("foo").await.unwrap().is_none());
1817    }
1818
1819    #[tokio::test]
1820    async fn delete_nonexistent_database() {
1821        let dir = tempdir().unwrap();
1822        let store = open_directory_store(dir.path());
1823        assert!(!store.delete("foo").await.unwrap());
1824    }
1825
1826    #[tokio::test]
1827    async fn delete_graph() {
1828        let dir = tempdir().unwrap();
1829        let store = open_directory_store(dir.path());
1830        let graph = store.create("foo").await.unwrap();
1831        assert!(store.open("foo").await.unwrap().is_some());
1832        graph.delete().await.unwrap();
1833        assert!(store.open("foo").await.unwrap().is_none());
1834    }
1835
1836    #[tokio::test]
1837    async fn recreate_graph() {
1838        let dir = tempdir().unwrap();
1839        let store = open_directory_store(dir.path());
1840        let graph = store.create("foo").await.unwrap();
1841        let builder = store.create_base_layer().await.unwrap();
1842        let layer = builder.commit().await.unwrap();
1843        graph.set_head(&layer).await.unwrap();
1844        assert!(graph.head().await.unwrap().is_some());
1845        graph.delete().await.unwrap();
1846        store.create("foo").await.unwrap();
1847        assert!(graph.head().await.unwrap().is_none());
1848    }
1849
1850    #[tokio::test]
1851    async fn list_databases() {
1852        let dir = tempdir().unwrap();
1853        let store = open_directory_store(dir.path());
1854        assert!(store.labels().await.unwrap().is_empty());
1855        let _ = store.create("foo").await.unwrap();
1856        let one = vec!["foo".to_string()];
1857        assert_eq!(store.labels().await.unwrap(), one);
1858        let _ = store.create("bar").await.unwrap();
1859        let two = vec!["bar".to_string(), "foo".to_string()];
1860        let mut left = store.labels().await.unwrap();
1861        left.sort();
1862        assert_eq!(left, two);
1863    }
1864}