Implement persist sync feature

Cargo.toml

@@ -19,18 +19,19 @@ perf_measurements = ["dep:performance_measurement", "dep:performance_measurement
 eyre = "0.6.12"
 derive_more = { version = "1.0.0", features = ["from", "error", "display", "into"] }
 tokio = { version = "1", features = ["full"] }
+tracing = "0.1.40"
 rkyv = { version = "0.8.9", features = ["uuid-1"] }
 lockfree = { version = "0.5.1" }
 worktable_codegen = { path = "codegen", version = "0.4.1" }
 scc = "2.1.16"
 futures = "0.3.30"
 uuid = { version = "1.10.0", features = ["v4"] }
 data_bucket = { git = "https://github.com/pathscale/DataBucket", branch = "main" }
-#data_bucket = { path = "../DataBucket", version = "0.1.1" }
+# data_bucket = { path = "../DataBucket", version = "0.1.0" }
 performance_measurement_codegen = { path = "performance_measurement/codegen", version = "0.1.0", optional = true }
 performance_measurement = { path = "performance_measurement", version = "0.1.0", optional = true }
-indexset = { version = "0.10.3", features = ["concurrent", "multimap"] }
-# indexset = { path = "../indexset", version = "0.10.3", features = ["concurrent", "multimap"] }
+indexset = { version = "0.11.2", features = ["concurrent", "cdc", "multimap"] }
+convert_case = "0.6.0"
 
 [dev-dependencies]
 async-std = "1.13.0"

codegen/src/persist_index/mod.rs

@@ -6,6 +6,7 @@ use crate::persist_index::parser::Parser;
 
 mod generator;
 mod parser;
+mod space;
 
 pub fn expand(input: TokenStream) -> syn::Result<TokenStream> {
     let input_struct = Parser::parse_struct(input)?;
@@ -14,19 +15,20 @@ pub fn expand(input: TokenStream) -> syn::Result<TokenStream> {
     let type_def = gen.gen_persist_type()?;
     let persistable_def = gen.gen_persistable_impl()?;
     let impl_def = gen.gen_persist_impl()?;
+    let space_index = gen.gen_space_index();
 
     Ok(quote! {
         #type_def
         #impl_def
         #persistable_def
+        #space_index
     })
 }
 
 #[cfg(test)]
 mod tests {
     use quote::quote;
     use rkyv::{Archive, Deserialize, Serialize};
-    use scc::TreeIndex;
 
     use crate::persist_index::expand;
 
@@ -39,11 +41,6 @@ mod tests {
         pub length: u32,
     }
 
-    pub struct TestIndex {
-        test_idx: TreeIndex<i64, Link>,
-        exchange_idx: TreeIndex<String, std::sync::Arc<lockfree::set::Set<Link>>>,
-    }
-
     #[test]
     fn test() {
         let input = quote! {

codegen/src/persist_index/parser.rs

@@ -12,3 +12,41 @@ impl Parser {
         }
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use crate::persist_index::parser::Parser;
+    use quote::quote;
+
+    #[test]
+    fn parses_index_struct() {
+        let input = quote! {
+            #[derive(Debug, Default, Clone)]
+            pub struct TestIndex {
+                test_idx: TreeIndex<i64, Link>,
+                exchnage_idx: TreeIndex<String, std::sync::Arc<LockFreeSet<Link>>>
+            }
+        };
+        assert!(Parser::parse_struct(input).is_ok())
+    }
+
+    #[test]
+    fn errors_on_type() {
+        let input = quote! {
+            pub type TestIndex = Srting;
+        };
+        assert!(Parser::parse_struct(input).is_err())
+    }
+
+    #[test]
+    fn errors_on_enum() {
+        let input = quote! {
+            #[derive(Debug, Default, Clone)]
+            pub enum TestIndex {
+                First,
+                Second
+            }
+        };
+        assert!(Parser::parse_struct(input).is_err())
+    }
+}

codegen/src/persist_index/space.rs

@@ -0,0 +1,129 @@
+use proc_macro2::{Literal, TokenStream};
+use quote::quote;
+
+use crate::name_generator::WorktableNameGenerator;
+use crate::persist_index::generator::Generator;
+
+impl Generator {
+    pub fn gen_space_index(&self) -> TokenStream {
+        let secondary_index = self.gen_space_secondary_index_type();
+        let secondary_impl = self.gen_space_secondary_index_impl_space_index();
+        let secondary_index_events = self.gen_space_secondary_index_events_type();
+
+        quote! {
+            #secondary_index_events
+            #secondary_index
+            #secondary_impl
+        }
+    }
+
+    fn gen_space_secondary_index_events_type(&self) -> TokenStream {
+        let name_generator = WorktableNameGenerator::from_index_ident(&self.struct_def.ident);
+        let ident = name_generator.get_space_secondary_index_events_ident();
+
+        let fields: Vec<_> = self
+            .field_types
+            .iter()
+            .map(|(i, t)| {
+                quote! {
+                    #i: Vec<indexset::cdc::change::ChangeEvent<
+                        indexset::core::pair::Pair<#t, Link>
+                    >>,
+                }
+            })
+            .collect();
+
+        quote! {
+            #[derive(Debug)]
+            pub struct #ident {
+                #(#fields)*
+            }
+        }
+    }
+
+    fn gen_space_secondary_index_type(&self) -> TokenStream {
+        let name_generator = WorktableNameGenerator::from_index_ident(&self.struct_def.ident);
+        let ident = name_generator.get_space_secondary_index_ident();
+        let inner_const_name = name_generator.get_page_inner_size_const_ident();
+
+        let fields: Vec<_> = self
+            .field_types
+            .iter()
+            .map(|(i, t)| {
+                quote! {
+                    #i: SpaceIndex<#t, { #inner_const_name as u32}>,
+                }
+            })
+            .collect();
+
+        quote! {
+            #[derive(Debug)]
+            pub struct #ident {
+                #(#fields)*
+            }
+        }
+    }
+
+    fn gen_space_secondary_index_impl_space_index(&self) -> TokenStream {
+        let name_generator = WorktableNameGenerator::from_index_ident(&self.struct_def.ident);
+        let events_ident = name_generator.get_space_secondary_index_events_ident();
+        let ident = name_generator.get_space_secondary_index_ident();
+
+        let from_table_files_path_fn = self.gen_space_secondary_index_from_table_files_path_fn();
+        let index_process_change_events_fn =
+            self.gen_space_secondary_index_process_change_events_fn();
+
+        quote! {
+            impl SpaceSecondaryIndexOps<#events_ident> for #ident {
+                #from_table_files_path_fn
+                #index_process_change_events_fn
+            }
+        }
+    }
+
+    fn gen_space_secondary_index_from_table_files_path_fn(&self) -> TokenStream {
+        let fields: Vec<_> = self
+            .field_types
+            .keys()
+            .map(|i| {
+                let literal_name = Literal::string(i.to_string().as_str());
+                quote! {
+                    #i: SpaceIndex::secondary_from_table_files_path(path, #literal_name)?,
+                }
+            })
+            .collect();
+
+        quote! {
+            fn from_table_files_path<S: AsRef<str>>(path: S) -> eyre::Result<Self> {
+                let path = path.as_ref();
+                Ok(Self {
+                    #(#fields)*
+                })
+            }
+        }
+    }
+
+    fn gen_space_secondary_index_process_change_events_fn(&self) -> TokenStream {
+        let name_generator = WorktableNameGenerator::from_index_ident(&self.struct_def.ident);
+        let events_ident = name_generator.get_space_secondary_index_events_ident();
+
+        let process: Vec<_> = self
+            .field_types
+            .keys()
+            .map(|i| {
+                quote! {
+                    for event in events.#i {
+                        self.#i.process_change_event(event)?;
+                    }
+                }
+            })
+            .collect();
+
+        quote! {
+            fn process_change_events(&mut self, events: #events_ident) -> eyre::Result<()> {
+                #(#process)*
+                core::result::Result::Ok(())
+            }
+        }
+    }
+}

codegen/src/persist_table/generator/mod.rs

@@ -7,6 +7,7 @@ use crate::name_generator::WorktableNameGenerator;
 pub use space_file::WT_INDEX_EXTENSION;
 
 mod size_measurable;
+mod space;
 mod space_file;
 
 pub struct Generator {
@@ -22,11 +23,21 @@ impl WorktableNameGenerator {
         )
     }
 
-    //pub fn get_space_ident(&self) -> Ident {
-    //    Ident::new(format!("{}Space", self.name).as_str(), Span::mixed_site())
-    //}
-
     pub fn get_dir_name(&self) -> String {
         self.name.from_case(Case::Pascal).to_case(Case::Snake)
     }
+
+    pub fn get_persistence_engine_ident(&self) -> Ident {
+        Ident::new(
+            format!("{}PersistenceEngine", self.name).as_str(),
+            Span::mixed_site(),
+        )
+    }
+
+    pub fn get_persistence_task_ident(&self) -> Ident {
+        Ident::new(
+            format!("{}PersistenceTask", self.name).as_str(),
+            Span::mixed_site(),
+        )
+    }
 }

codegen/src/persist_table/generator/space.rs

@@ -0,0 +1,45 @@
+use crate::name_generator::WorktableNameGenerator;
+use crate::persist_table::generator::Generator;
+use proc_macro2::TokenStream;
+use quote::quote;
+
+impl Generator {
+    pub fn get_persistence_task_type(&self) -> TokenStream {
+        let name_generator = WorktableNameGenerator::from_struct_ident(&self.struct_def.ident);
+        let ident = name_generator.get_persistence_task_ident();
+        let primary_key_type = name_generator.get_primary_key_type_ident();
+        let space_secondary_indexes_events =
+            name_generator.get_space_secondary_index_events_ident();
+
+        quote! {
+            pub type #ident = PersistenceTask<
+                <<#primary_key_type as TablePrimaryKey>::Generator as PrimaryKeyGeneratorState>::State,
+                #primary_key_type,
+                #space_secondary_indexes_events,
+            >;
+        }
+    }
+
+    pub fn get_persistence_engine_type(&self) -> TokenStream {
+        let name_generator = WorktableNameGenerator::from_struct_ident(&self.struct_def.ident);
+        let ident = name_generator.get_persistence_engine_ident();
+        let primary_key_type = name_generator.get_primary_key_type_ident();
+        let inner_const_name = name_generator.get_page_inner_size_const_ident();
+        let space_secondary_indexes = name_generator.get_space_secondary_index_ident();
+        let space_secondary_indexes_events =
+            name_generator.get_space_secondary_index_events_ident();
+
+        quote! {
+            pub type #ident = PersistenceEngine<
+                SpaceData<
+                    <<#primary_key_type as TablePrimaryKey>::Generator as PrimaryKeyGeneratorState>::State,
+                    { #inner_const_name as u32 }
+                >,
+                SpaceIndex<#primary_key_type, { #inner_const_name as u32 }>,
+                #space_secondary_indexes,
+                #primary_key_type,
+                #space_secondary_indexes_events,
+            >;
+        }
+    }
+}