Typed dicts

Cargo.toml

@@ -25,6 +25,13 @@ flate2 = "1.0"
 rayon = "1.4"
 thiserror = "1.0"
 async-trait = "0.1"
+itertools = "0.10"
+rug = {version="1.16", default-features=false, features=["integer","rational"]}
+num-derive = "0.3"
+num-traits = "0.2"
+chrono = "0.4"
+base64 = "0.13"
+hex = "0.4"
 
 [dev-dependencies]
 tempfile = "3.1"

benches/bench.rs

@@ -3,7 +3,7 @@ extern crate test;
 
 use tempfile::tempdir;
 use terminus_store;
-use terminus_store::layer::StringTriple;
+use terminus_store::layer::ValueTriple;
 use test::Bencher;
 
 #[bench]
@@ -14,7 +14,7 @@ fn bench_add_string_triple(b: &mut Bencher) {
     let mut count = 1;
     b.iter(|| {
         layer_builder
-            .add_string_triple(StringTriple::new_value(
+            .add_value_triple(ValueTriple::new_string_value(
                 &count.to_string(),
                 &count.to_string(),
                 &count.to_string(),

benches/builder/data.rs

@@ -1,7 +1,7 @@
 use rand::distributions::Alphanumeric;
 use rand::prelude::*;
 use std::iter;
-use terminus_store::layer::StringTriple;
+use terminus_store::layer::ValueTriple;
 
 fn random_string<R: Rng>(rand: &mut R, len_min: usize, len_max: usize) -> String {
     let len: usize = rand.gen_range(len_min..len_max);
@@ -50,19 +50,19 @@ impl<R: Rng> TestData<R> {
         }
     }
 
-    pub fn random_triple(&mut self) -> StringTriple {
+    pub fn random_triple(&mut self) -> ValueTriple {
         let subject_ix = self.rand.gen_range(0..self.nodes.len());
         let predicate_ix = self.rand.gen_range(0..self.predicates.len());
         if self.rand.gen() {
             let object_ix = self.rand.gen_range(0..self.nodes.len());
-            StringTriple::new_node(
+            ValueTriple::new_node(
                 &self.nodes[subject_ix],
                 &self.predicates[predicate_ix],
                 &self.nodes[object_ix],
             )
         } else {
             let object_ix = self.rand.gen_range(0..self.values.len());
-            StringTriple::new_value(
+            ValueTriple::new_string_value(
                 &self.nodes[subject_ix],
                 &self.predicates[predicate_ix],
                 &self.values[object_ix],

benches/builder/main.rs

@@ -38,7 +38,7 @@ fn build_base_layer_1000(b: &mut Bencher) {
         let builder = store.create_base_layer().unwrap();
 
         for triple in triples.iter() {
-            builder.add_string_triple(triple.clone()).unwrap();
+            builder.add_value_triple(triple.clone()).unwrap();
         }
 
         let _base_layer = builder.commit().unwrap();
@@ -78,7 +78,7 @@ fn build_nonempty_child_layer_on_empty_base_layer(b: &mut Bencher) {
         let builder = base_layer.open_write().unwrap();
 
         for triple in triples.iter() {
-            builder.add_string_triple(triple.clone()).unwrap();
+            builder.add_value_triple(triple.clone()).unwrap();
         }
 
         builder.commit().unwrap();
@@ -97,7 +97,7 @@ fn build_nonempty_child_layer_on_nonempty_base_layer(b: &mut Bencher) {
     let builder = store.create_base_layer().unwrap();
 
     for _ in 0..1000 {
-        builder.add_string_triple(data.random_triple()).unwrap();
+        builder.add_value_triple(data.random_triple()).unwrap();
     }
     let base_layer = builder.commit().unwrap();
 
@@ -110,7 +110,7 @@ fn build_nonempty_child_layer_on_nonempty_base_layer(b: &mut Bencher) {
         let builder = base_layer.open_write().unwrap();
 
         for triple in triples.iter() {
-            builder.add_string_triple(triple.clone()).unwrap();
+            builder.add_value_triple(triple.clone()).unwrap();
         }
 
         builder.commit().unwrap();

examples/print_graph.rs

@@ -1,6 +1,7 @@
 use std::env;
 
 use std::io;
+use terminus_store::structure::TdbDataType;
 use terminus_store::*;
 use tokio;
 
@@ -21,15 +22,15 @@ async fn print_graph(store_path: &str, graph: &str) -> io::Result<()> {
                     .expect("expected id triple to be mapable to string");
 
                 println!(
-                    "{}, {}, {} {}",
+                    "{}, {}, {} {:?}",
                     triple.subject,
                     triple.predicate,
                     match triple.object {
                         ObjectType::Node(_) => "node",
                         ObjectType::Value(_) => "value",
                     },
                     match triple.object {
-                        ObjectType::Node(n) => n,
+                        ObjectType::Node(n) => String::make_entry(&n),
                         ObjectType::Value(v) => v,
                     }
                 );

examples/write_to_graph.rs

@@ -7,8 +7,8 @@ use tokio;
 use tokio::io::{self, AsyncBufReadExt};
 
 enum Command {
-    Add(StringTriple),
-    Remove(StringTriple),
+    Add(ValueTriple),
+    Remove(ValueTriple),
 }
 
 async fn parse_command(s: &str) -> io::Result<Command> {
@@ -25,8 +25,8 @@ async fn parse_command(s: &str) -> io::Result<Command> {
         let object = &matches[5];
 
         let triple = match object_type_name {
-            "node" => StringTriple::new_node(subject, predicate, object),
-            "value" => StringTriple::new_value(subject, predicate, object),
+            "node" => ValueTriple::new_node(subject, predicate, object),
+            "value" => ValueTriple::new_string_value(subject, predicate, object),
             _ => {
                 return Err(io::Error::new(
                     io::ErrorKind::InvalidData,
@@ -84,8 +84,8 @@ async fn process_commands(store_path: &str, graph: &str) -> io::Result<()> {
         // Since no io is happening, adding triples to the builder is
         // not a future.
         match command {
-            Command::Add(triple) => builder.add_string_triple(triple)?,
-            Command::Remove(triple) => builder.remove_string_triple(triple)?,
+            Command::Add(triple) => builder.add_value_triple(triple)?,
+            Command::Remove(triple) => builder.remove_value_triple(triple)?,
         }
     }
 

src/layer/builder.rs

@@ -1,5 +1,6 @@
 use std::io;
 
+use bytes::{Bytes, BytesMut};
 use futures::stream::TryStreamExt;
 use rayon::prelude::*;
 
@@ -8,32 +9,35 @@ use crate::storage::*;
 use crate::structure::util;
 use crate::structure::*;
 
-pub struct DictionarySetFileBuilder<F: 'static + FileStore> {
-    node_dictionary_builder: PfcDictFileBuilder<F::Write>,
-    predicate_dictionary_builder: PfcDictFileBuilder<F::Write>,
-    value_dictionary_builder: PfcDictFileBuilder<F::Write>,
+pub struct DictionarySetFileBuilder<F: 'static + FileLoad + FileStore> {
+    node_files: DictionaryFiles<F>,
+    predicate_files: DictionaryFiles<F>,
+    value_files: TypedDictionaryFiles<F>,
+    node_dictionary_builder: StringDictBufBuilder<BytesMut, BytesMut>,
+    predicate_dictionary_builder: StringDictBufBuilder<BytesMut, BytesMut>,
+    value_dictionary_builder: TypedDictBufBuilder<BytesMut, BytesMut, BytesMut, BytesMut>,
 }
 
 impl<F: 'static + FileLoad + FileStore> DictionarySetFileBuilder<F> {
     pub async fn from_files(
         node_files: DictionaryFiles<F>,
         predicate_files: DictionaryFiles<F>,
-        value_files: DictionaryFiles<F>,
+        value_files: TypedDictionaryFiles<F>,
     ) -> io::Result<Self> {
-        let node_dictionary_builder = PfcDictFileBuilder::new(
-            node_files.blocks_file.open_write().await?,
-            node_files.offsets_file.open_write().await?,
-        );
-        let predicate_dictionary_builder = PfcDictFileBuilder::new(
-            predicate_files.blocks_file.open_write().await?,
-            predicate_files.offsets_file.open_write().await?,
-        );
-        let value_dictionary_builder = PfcDictFileBuilder::new(
-            value_files.blocks_file.open_write().await?,
-            value_files.offsets_file.open_write().await?,
+        let node_dictionary_builder = StringDictBufBuilder::new(BytesMut::new(), BytesMut::new());
+        let predicate_dictionary_builder =
+            StringDictBufBuilder::new(BytesMut::new(), BytesMut::new());
+        let value_dictionary_builder = TypedDictBufBuilder::new(
+            BytesMut::new(),
+            BytesMut::new(),
+            BytesMut::new(),
+            BytesMut::new(),
         );
 
         Ok(Self {
+            node_files,
+            predicate_files,
+            value_files,
             node_dictionary_builder,
             predicate_dictionary_builder,
             value_dictionary_builder,
@@ -43,91 +47,117 @@ impl<F: 'static + FileLoad + FileStore> DictionarySetFileBuilder<F> {
     /// Add a node string.
     ///
     /// Panics if the given node string is not a lexical successor of the previous node string.
-    pub async fn add_node(&mut self, node: &str) -> io::Result<u64> {
-        let id = self.node_dictionary_builder.add(node).await?;
+    pub fn add_node(&mut self, node: &str) -> u64 {
+        let id = self
+            .node_dictionary_builder
+            .add(Bytes::copy_from_slice(node.as_bytes()));
 
-        Ok(id)
+        id
     }
 
     /// Add a predicate string.
     ///
     /// Panics if the given predicate string is not a lexical successor of the previous node string.
-    pub async fn add_predicate(&mut self, predicate: &str) -> io::Result<u64> {
-        let id = self.predicate_dictionary_builder.add(predicate).await?;
+    pub fn add_predicate(&mut self, predicate: &str) -> u64 {
+        let id = self
+            .predicate_dictionary_builder
+            .add(Bytes::copy_from_slice(predicate.as_bytes()));
 
-        Ok(id)
+        id
     }
 
     /// Add a value string.
     ///
     /// Panics if the given value string is not a lexical successor of the previous value string.
-    pub async fn add_value(&mut self, value: &str) -> io::Result<u64> {
-        let id = self.value_dictionary_builder.add(value).await?;
+    pub fn add_value(&mut self, value: TypedDictEntry) -> u64 {
+        let id = self.value_dictionary_builder.add(value);
 
-        Ok(id)
+        id
     }
 
     /// Add nodes from an iterable.
     ///
     /// Panics if the nodes are not in lexical order, or if previous added nodes are a lexical succesor of any of these nodes.
-    pub async fn add_nodes<I: 'static + IntoIterator<Item = String> + Unpin + Send + Sync>(
+    pub fn add_nodes<I: 'static + IntoIterator<Item = String> + Unpin + Send + Sync>(
         &mut self,
         nodes: I,
-    ) -> io::Result<Vec<u64>>
+    ) -> Vec<u64>
     where
         <I as std::iter::IntoIterator>::IntoIter: Unpin + Send + Sync,
     {
         let mut ids = Vec::new();
         for node in nodes {
-            let id = self.add_node(&node).await?;
+            let id = self.add_node(&node);
             ids.push(id);
         }
 
-        Ok(ids)
+        ids
     }
 
     /// Add predicates from an iterable.
     ///
     /// Panics if the predicates are not in lexical order, or if previous added predicates are a lexical succesor of any of these predicates.
-    pub async fn add_predicates<I: 'static + IntoIterator<Item = String> + Unpin + Send + Sync>(
+    pub fn add_predicates<I: 'static + IntoIterator<Item = String> + Unpin + Send + Sync>(
         &mut self,
         predicates: I,
-    ) -> io::Result<Vec<u64>>
+    ) -> Vec<u64>
     where
         <I as std::iter::IntoIterator>::IntoIter: Unpin + Send + Sync,
     {
         let mut ids = Vec::new();
         for predicate in predicates {
-            let id = self.add_predicate(&predicate).await?;
+            let id = self.add_predicate(&predicate);
             ids.push(id);
         }
 
-        Ok(ids)
+        ids
     }
 
     /// Add values from an iterable.
     ///
     /// Panics if the values are not in lexical order, or if previous added values are a lexical succesor of any of these values.
-    pub async fn add_values<I: 'static + IntoIterator<Item = String> + Unpin + Send + Sync>(
+    pub fn add_values<I: 'static + IntoIterator<Item = TypedDictEntry> + Unpin + Send + Sync>(
         &mut self,
         values: I,
-    ) -> io::Result<Vec<u64>>
+    ) -> Vec<u64>
     where
         <I as std::iter::IntoIterator>::IntoIter: Unpin + Send + Sync,
     {
         let mut ids = Vec::new();
         for value in values {
-            let id = self.add_value(&value).await?;
+            let id = self.add_value(value);
             ids.push(id);
         }
 
-        Ok(ids)
+        ids
     }
 
     pub async fn finalize(self) -> io::Result<()> {
-        self.node_dictionary_builder.finalize().await?;
-        self.predicate_dictionary_builder.finalize().await?;
-        self.value_dictionary_builder.finalize().await?;
+        let (mut node_offsets_buf, mut node_data_buf) = self.node_dictionary_builder.finalize();
+        let (mut predicate_offsets_buf, mut predicate_data_buf) =
+            self.predicate_dictionary_builder.finalize();
+        let (
+            mut value_types_present_buf,
+            mut value_type_offsets_buf,
+            mut value_offsets_buf,
+            mut value_data_buf,
+        ) = self.value_dictionary_builder.finalize();
+
+        self.node_files
+            .write_all_from_bufs(&mut node_data_buf, &mut node_offsets_buf)
+            .await?;
+        self.predicate_files
+            .write_all_from_bufs(&mut predicate_data_buf, &mut predicate_offsets_buf)
+            .await?;
+
+        self.value_files
+            .write_all_from_bufs(
+                &mut value_types_present_buf,
+                &mut value_type_offsets_buf,
+                &mut value_offsets_buf,
+                &mut value_data_buf,
+            )
+            .await?;
 
         Ok(())
     }