#529 WIP propvalsub index

Author: joepio

lib/src/atoms.rs

@@ -3,7 +3,6 @@
 use crate::{errors::AtomicResult, values::Value};
 
 /// The Atom is the (non-validated) string representation of a piece of data.
-/// It's RichAtom sibling provides some extra methods.
 #[derive(Clone, Debug)]
 pub struct Atom {
     /// The URL where the resource is located

lib/src/commit.rs

@@ -352,10 +352,14 @@ impl Commit {
 
         if update_index {
             for atom in remove_atoms {
-                store.remove_atom_from_index(&atom, &resource_unedited)?;
+                store
+                    .remove_atom_from_index(&atom, &resource_unedited)
+                    .map_err(|e| format!("Error removing atom from index: {e}  Atom: {e}"))?
             }
             for atom in add_atoms {
-                store.add_atom_to_index(&atom, &resource)?;
+                store
+                    .add_atom_to_index(&atom, &resource)
+                    .map_err(|e| format!("Error adding atom to index: {e}  Atom: {e}"))?;
             }
         }
         Ok(resource)

lib/src/db.rs

@@ -1,6 +1,13 @@
 //! Persistent, ACID compliant, threadsafe to-disk store.
 //! Powered by Sled - an embedded database.
 
+mod migrations;
+mod prop_val_sub_index;
+mod query_index;
+mod reference_index;
+#[cfg(test)]
+pub mod test;
+
 use std::{
     collections::{HashMap, HashSet},
     sync::{Arc, Mutex},
@@ -10,6 +17,7 @@ use tracing::{instrument, trace};
 
 use crate::{
     commit::CommitResponse,
+    db::reference_index::key_to_atom,
     endpoints::{default_endpoints, Endpoint},
     errors::{AtomicError, AtomicResult},
     resources::PropVals,
@@ -19,20 +27,17 @@ use crate::{
 
 use self::{
     migrations::migrate_maybe,
+    prop_val_sub_index::add_atom_to_prop_val_sub_index,
     query_index::{
         atom_to_indexable_atoms, check_if_atom_matches_watched_query_filters, query_indexed,
-        update_indexed_member, watch_collection, IndexAtom, QueryFilter, END_CHAR,
+        update_indexed_member, watch_collection, QueryFilter, END_CHAR,
     },
+    reference_index::{add_atom_to_reference_index, delete_atom_from_reference_index},
 };
 
 // A function called by the Store when a Commit is accepted
 type HandleCommit = Box<dyn Fn(&CommitResponse) + Send + Sync>;
 
-mod migrations;
-mod query_index;
-#[cfg(test)]
-pub mod test;
-
 /// Inside the reference_index, each value is mapped to this type.
 /// The String on the left represents a Property URL, and the second one is the set of subjects.
 pub type PropSubjectMap = HashMap<String, HashSet<String>>;
@@ -54,13 +59,15 @@ pub struct Db {
     default_agent: Arc<Mutex<Option<crate::agents::Agent>>>,
     /// Stores all resources. The Key is the Subject as a `string.as_bytes()`, the value a [PropVals]. Propvals must be serialized using [bincode].
     resources: sled::Tree,
-    /// Index for all AtomicURLs, indexed by their Value. Used to speed up TPF queries. See [key_for_reference_index]
+    /// Index of all Atoms, sorted by {Value}-{Property}-{Subject}.
+    /// See [reference_index]
     reference_index: sled::Tree,
+    /// Index sorted by property + value.
+    /// Used for TPF queries where the property is known.
+    prop_val_sub_index: sled::Tree,
     /// Stores the members of Collections, easily sortable.
-    /// See [collections_index]
-    members_index: sled::Tree,
-    /// A list of all the Collections currently being used. Is used to update `members_index`.
-    /// See [collections_index]
+    query_index: sled::Tree,
+    /// A list of all the Collections currently being used. Is used to update `query_index`.
     watched_queries: sled::Tree,
     /// The address where the db will be hosted, e.g. http://localhost/
     server_url: String,
@@ -78,14 +85,16 @@ impl Db {
         let db = sled::open(path).map_err(|e|format!("Failed opening DB at this location: {:?} . Is another instance of Atomic Server running? {}", path, e))?;
         let resources = db.open_tree("resources_v1").map_err(|e|format!("Failed building resources. Your DB might be corrupt. Go back to a previous version and export your data. {}", e))?;
         let reference_index = db.open_tree("reference_index")?;
-        let members_index = db.open_tree("members_index")?;
+        let query_index = db.open_tree("members_index")?;
+        let prop_val_sub_index = db.open_tree("prop_val_sub_index")?;
         let watched_queries = db.open_tree("watched_queries")?;
         let store = Db {
             db,
             default_agent: Arc::new(Mutex::new(None)),
             resources,
             reference_index,
-            members_index,
+            query_index,
+            prop_val_sub_index,
             server_url,
             watched_queries,
             endpoints: default_endpoints(),
@@ -160,7 +169,7 @@ impl Db {
     /// Removes all values from the indexes.
     pub fn clear_index(&self) -> AtomicResult<()> {
         self.reference_index.clear()?;
-        self.members_index.clear()?;
+        self.query_index.clear()?;
         self.watched_queries.clear()?;
         Ok(())
     }
@@ -217,8 +226,8 @@ impl Storelike for Db {
     #[instrument(skip(self))]
     fn add_atom_to_index(&self, atom: &Atom, resource: &Resource) -> AtomicResult<()> {
         for index_atom in atom_to_indexable_atoms(atom)? {
-            // It's OK if this overwrites a value
             add_atom_to_reference_index(&index_atom, self)?;
+            add_atom_to_prop_val_sub_index(&index_atom, self)?;
             // Also update the query index to keep collections performant
             check_if_atom_matches_watched_query_filters(self, &index_atom, atom, false, resource)
                 .map_err(|e| {
@@ -697,42 +706,8 @@ impl Storelike for Db {
     }
 }
 
-#[instrument(skip(store))]
-fn add_atom_to_reference_index(index_atom: &IndexAtom, store: &Db) -> AtomicResult<()> {
-    let _existing = store
-        .reference_index
-        .insert(key_for_reference_index(index_atom).as_bytes(), b"")?;
-    Ok(())
-}
-
-#[instrument(skip(store))]
-fn delete_atom_from_reference_index(index_atom: &IndexAtom, store: &Db) -> AtomicResult<()> {
-    store
-        .reference_index
-        .remove(key_for_reference_index(index_atom).as_bytes())?;
-    Ok(())
-}
-
-/// Constructs the Key for the index_value cache.
-fn key_for_reference_index(atom: &IndexAtom) -> String {
-    format!("{}\n{}\n{}", atom.value, atom.property, atom.subject)
-}
-
-/// Parses a Value index key string, converts it into an atom. Note that the Value of the atom will allways be a single AtomicURL here.
-fn key_to_atom(key: &str) -> AtomicResult<Atom> {
-    let mut parts = key.split('\n');
-    let val = parts.next().ok_or("Invalid key for value index")?;
-    let prop = parts.next().ok_or("Invalid key for value index")?;
-    let subj = parts.next().ok_or("Invalid key for value index")?;
-    Ok(Atom::new(
-        subj.into(),
-        prop.into(),
-        Value::AtomicUrl(val.into()),
-    ))
-}
-
 fn corrupt_db_message(subject: &str) -> String {
-    format!("Could not deserialize item {} from database. DB is possibly corrupt, could be due to an update or a lack of migrations. Restore to a previous version, export / serialize your data and import your data again.", subject)
+    format!("Could not deserialize item {} from database. DB is possibly corrupt, could be due to an update or a lack of migrations. Restore to a previous version, export your data and import your data again.", subject)
 }
 
-const DB_CORRUPT_MSG: &str = "Could not deserialize item from database. DB is possibly corrupt, could be due to an update or a lack of migrations. Restore to a previous version, export / serialize your data and import your data again.";
+const DB_CORRUPT_MSG: &str = "Could not deserialize item from database. DB is possibly corrupt, could be due to an update or a lack of migrations. Restore to a previous version, export your data and import your data again.";

lib/src/db/prop_val_sub_index.rs

@@ -0,0 +1,76 @@
+//! Index sorted by {Property}-{Value}-{Subject}.
+
+use tracing::instrument;
+
+use crate::{errors::AtomicResult, Db};
+
+use super::query_index::{IndexAtom, SEPARATION_BIT};
+
+/// Finds all Atoms for a given {property}-{value} tuple.
+pub fn find_in_prop_val_sub_index(
+    store: &Db,
+    prop: &str,
+    val: Option<&str>,
+) -> AtomicResult<sled::Iter> {
+    let mut prefix: Vec<u8> = [prop.as_bytes(), &[SEPARATION_BIT]].concat();
+    if let Some(value) = val {
+        prefix.extend(value.as_bytes());
+        prefix.extend(&[SEPARATION_BIT]);
+    }
+    Ok(store.prop_val_sub_index.scan_prefix(prefix))
+}
+
+#[instrument(skip(store))]
+pub fn add_atom_to_prop_val_sub_index(index_atom: &IndexAtom, store: &Db) -> AtomicResult<()> {
+    let _existing = store
+        .prop_val_sub_index
+        .insert(key_from_atom(index_atom), b"")?;
+    Ok(())
+}
+
+/// Constructs the Key for the prop_val_sub_index.
+fn key_from_atom(atom: &IndexAtom) -> Vec<u8> {
+    [
+        atom.property.as_bytes(),
+        &[SEPARATION_BIT],
+        atom.value.as_bytes(),
+        &[SEPARATION_BIT],
+        atom.subject.as_bytes(),
+    ]
+    .concat()
+}
+
+/// Parses a Value index key string, converts it into an atom.
+/// Note that the Value of the atom will always be a single AtomicURL here.
+fn key_to_index_atom(key: &[u8]) -> AtomicResult<IndexAtom> {
+    let mut parts = key.split(|b| b == &SEPARATION_BIT);
+    let prop = std::str::from_utf8(parts.next().ok_or("Invalid key for prop_val_sub_index")?)
+        .map_err(|_| "Can't parse prop into string")?;
+    let val = std::str::from_utf8(parts.next().ok_or("Invalid key for prop_val_sub_index")?)
+        .map_err(|_| "Can't parse val into string")?;
+    let sub = std::str::from_utf8(parts.next().ok_or("Invalid key for prop_val_sub_index")?)
+        .map_err(|_| "Can't parse subject into string")?;
+    Ok(IndexAtom {
+        property: prop.into(),
+        value: val.into(),
+        subject: sub.into(),
+    })
+}
+
+#[cfg(test)]
+mod test {
+    use super::*;
+    use crate::db::query_index::IndexAtom;
+
+    #[test]
+    fn round_trip() {
+        let atom = IndexAtom {
+            property: "http://example.com/prop".into(),
+            value: "http://example.com/val \n hello \n".into(),
+            subject: "http://example.com/subj".into(),
+        };
+        let key = key_from_atom(&atom);
+        let atom2 = key_to_index_atom(&key).unwrap();
+        assert_eq!(atom, atom2);
+    }
+}

lib/src/db/query_index.rs

@@ -1,5 +1,4 @@
-//! The Collections Cache is used to speed up queries.
-//! It sorts Members by their Value, so we can quickly paginate and sort.
+//! The QueryIndex is used to speed up queries by persisting filtered, sorted collections.
 //! It relies on lexicographic ordering of keys, which Sled utilizes using `scan_prefix` queries.
 
 use crate::{
@@ -13,6 +12,7 @@ use serde::{Deserialize, Serialize};
 /// A subset of a full [Query].
 /// Represents a sorted filter on the Store.
 /// A Value in the `watched_collections`.
+/// Used as keys in the query_index.
 /// These are used to check whether collections have to be updated when values have changed.
 #[derive(Debug, Clone, Serialize, Deserialize)]
 pub struct QueryFilter {
@@ -34,10 +34,10 @@ impl From<&Query> for QueryFilter {
     }
 }
 
-/// Differs from a Regular Atom, since the value here is always a string,
+/// Differs from a regular [Atom], since the value here is always a string,
 /// and in the case of ResourceArrays, only a _single_ subject is used for each atom.
 /// One IndexAtom for every member of the ResourceArray is created.
-#[derive(Debug, Clone)]
+#[derive(Debug, Clone, PartialEq, Eq)]
 pub struct IndexAtom {
     pub subject: String,
     pub property: String,
@@ -47,9 +47,12 @@ pub struct IndexAtom {
 /// Last character in lexicographic ordering
 pub const FIRST_CHAR: &str = "\u{0000}";
 pub const END_CHAR: &str = "\u{ffff}";
+/// We can only store one bytearray as a key in Sled.
+/// We separate the various items in it using this bit that's illegal in UTF-8.
+pub const SEPARATION_BIT: u8 = 0xff;
 
 #[tracing::instrument(skip(store))]
-/// Performs a query on the `members_index` Tree, which is a lexicographic sorted list of all hits for QueryFilters.
+/// Performs a query on the `query_index` Tree, which is a lexicographic sorted list of all hits for QueryFilters.
 pub fn query_indexed(store: &Db, q: &Query) -> AtomicResult<QueryResult> {
     // When there is no explicit start / end value passed, we use the very first and last
     // lexicographic characters in existence to make the range practically encompass all values.
@@ -68,9 +71,9 @@ pub fn query_indexed(store: &Db, q: &Query) -> AtomicResult<QueryResult> {
 
     let iter: Box<dyn Iterator<Item = std::result::Result<(sled::IVec, sled::IVec), sled::Error>>> =
         if q.sort_desc {
-            Box::new(store.members_index.range(start_key..end_key).rev())
+            Box::new(store.query_index.range(start_key..end_key).rev())
         } else {
-            Box::new(store.members_index.range(start_key..end_key))
+            Box::new(store.query_index.range(start_key..end_key))
         };
 
     let mut subjects: Vec<String> = vec![];
@@ -293,18 +296,14 @@ pub fn update_indexed_member(
         Some(subject),
     )?;
     if delete {
-        store.members_index.remove(key)?;
+        store.query_index.remove(key)?;
     } else {
-        store.members_index.insert(key, b"")?;
+        store.query_index.insert(key, b"")?;
     }
     Ok(())
 }
 
-/// We can only store one bytearray as a key in Sled.
-/// We separate the various items in it using this bit that's illegal in UTF-8.
-const SEPARATION_BIT: u8 = 0xff;
-
-/// Maximum string length for values in the members_index. Should be long enough to contain pretty long URLs, but not very long documents.
+/// Maximum string length for values in the query_index. Should be long enough to contain pretty long URLs, but not very long documents.
 pub const MAX_LEN: usize = 120;
 
 /// Creates a key for a collection + value combination.

lib/src/db/reference_index.rs

@@ -0,0 +1,43 @@
+//! The ReferenceIndex is for quickly finding incoming links.
+//! It indexes all values in the database.
+
+use super::query_index::IndexAtom;
+use crate::{errors::AtomicResult, Atom, Db, Value};
+use tracing::instrument;
+
+#[instrument(skip(store))]
+pub fn add_atom_to_reference_index(index_atom: &IndexAtom, store: &Db) -> AtomicResult<()> {
+    let _existing = store
+        .reference_index
+        .insert(key_from_atom(index_atom).as_bytes(), b"")?;
+    Ok(())
+}
+
+#[instrument(skip(store))]
+pub fn delete_atom_from_reference_index(index_atom: &IndexAtom, store: &Db) -> AtomicResult<()> {
+    store
+        .reference_index
+        .remove(key_from_atom(index_atom).as_bytes())?;
+    Ok(())
+}
+
+/// Constructs the Key for the index_value cache.
+fn key_from_atom(atom: &IndexAtom) -> String {
+    format!("{}\n{}\n{}", atom.value, atom.property, atom.subject)
+}
+
+/// Parses a Value index key string, converts it into an atom.
+/// Note that the Value of the atom will always be a single AtomicURL here.
+pub fn key_to_atom(key: &str) -> AtomicResult<Atom> {
+    let mut parts = key.split('\n');
+    let val = parts.next().ok_or(ERR_MSG)?;
+    let prop = parts.next().ok_or(ERR_MSG)?;
+    let subj = parts.next().ok_or(ERR_MSG)?;
+    Ok(Atom::new(
+        subj.into(),
+        prop.into(),
+        Value::AtomicUrl(val.into()),
+    ))
+}
+
+static ERR_MSG: &str = "Invalid key in value index. Try starting the server with --rebuild-index.";